lowest sequence number protocol

Author: somdoron

.specify/templates/plan-template.md

@@ -18,7 +18,7 @@
    → Update Progress Tracking: Initial Constitution Check
 5. Execute Phase 0 → research.md
    → If NEEDS CLARIFICATION remain: ERROR "Resolve unknowns"
-6. Execute Phase 1 → contracts, data-model.md, quickstart.md, agent-specific template file (e.g., `CLAUDE.md` for Claude Code, `.github/copilot-instructions.md` for GitHub Copilot, `GEMINI.md` for Gemini CLI, `QWEN.md` for Qwen Code or `AGENTS.md` for opencode).
+6. Execute Phase 1 → data-model.md, quickstart.md, agent-specific template file (e.g., `CLAUDE.md` for Claude Code, `.github/copilot-instructions.md` for GitHub Copilot, `GEMINI.md` for Gemini CLI, `QWEN.md` for Qwen Code or `AGENTS.md` for opencode).
 7. Re-evaluate Constitution Check section
    → If new violations: Refactor design, return to Phase 1
    → Update Progress Tracking: Post-Design Constitution Check
@@ -66,6 +66,7 @@
 - [ ] ZIO primitives used for all concurrent operations
 - [ ] ZStream used for streaming, no external streaming libraries
 - [ ] Resource management follows ZIO Scope patterns
+- [ ] `suiteAll` is used instead of `suite`
 
 ### V. Test-Driven Maintenance
 - [ ] Bug fixes include reproducing test cases
@@ -81,49 +82,9 @@ specs/[###-feature]/
 ├── research.md          # Phase 0 output (/plan command)
 ├── data-model.md        # Phase 1 output (/plan command)
 ├── quickstart.md        # Phase 1 output (/plan command)
-├── contracts/           # Phase 1 output (/plan command)
 └── tasks.md             # Phase 2 output (/tasks command - NOT created by /plan)
 ```
 
-### Source Code (repository root)
-```
-# Option 1: Single project (DEFAULT)
-src/
-├── models/
-├── services/
-├── cli/
-└── lib/
-
-tests/
-├── contract/
-├── integration/
-└── unit/
-
-# Option 2: Web application (when "frontend" + "backend" detected)
-backend/
-├── src/
-│   ├── models/
-│   ├── services/
-│   └── api/
-└── tests/
-
-frontend/
-├── src/
-│   ├── components/
-│   ├── pages/
-│   └── services/
-└── tests/
-
-# Option 3: Mobile + API (when "iOS/Android" detected)
-api/
-└── [same as backend above]
-
-ios/ or android/
-└── [platform-specific structure]
-```
-
-**Structure Decision**: [DEFAULT to Option 1 unless Technical Context indicates web/mobile app]
-
 ## Phase 0: Outline & Research
 1. **Extract unknowns from Technical Context** above:
    - For each NEEDS CLARIFICATION → research task
@@ -145,29 +106,19 @@ ios/ or android/
 
 **Output**: research.md with all NEEDS CLARIFICATION resolved
 
-## Phase 1: Design & Contracts
+## Phase 1: Design
 *Prerequisites: research.md complete*
 
 1. **Extract entities from feature spec** → `data-model.md`:
    - Entity name, fields, relationships
    - Validation rules from requirements
    - State transitions if applicable
 
-2. **Generate API contracts** from functional requirements:
-   - For each user action → endpoint
-   - Use standard REST/GraphQL patterns
-   - Output OpenAPI/GraphQL schema to `/contracts/`
-
-3. **Generate contract tests** from contracts:
-   - One test file per endpoint
-   - Assert request/response schemas
-   - Tests must fail (no implementation yet)
-
-4. **Extract test scenarios** from user stories:
+2. **Extract test scenarios** from user stories:
    - Each story → integration test scenario
    - Quickstart test = story validation steps
 
-5. **Update agent file incrementally** (O(1) operation):
+3. **Update agent file incrementally** (O(1) operation):
    - Run `.specify/scripts/bash/update-agent-context.sh cursor`
      **IMPORTANT**: Execute it exactly as specified above. Do not add or remove any arguments.
    - If exists: Add only NEW tech from current plan
@@ -176,15 +127,14 @@ ios/ or android/
    - Keep under 150 lines for token efficiency
    - Output to repository root
 
-**Output**: data-model.md, /contracts/*, failing tests, quickstart.md, agent-specific file
+**Output**: data-model.md, failing tests, quickstart.md, agent-specific file
 
 ## Phase 2: Task Planning Approach
 *This section describes what the /tasks command will do - DO NOT execute during /plan*
 
 **Task Generation Strategy**:
 - Load `.specify/templates/tasks-template.md` as base
-- Generate tasks from Phase 1 design docs (contracts, data model, quickstart)
-- Each contract → contract test task [P]
+- Generate tasks from Phase 1 design docs (data model, quickstart)
 - Each entity → model creation task [P] 
 - Each user story → integration test task
 - Implementation tasks to make tests pass

.specify/templates/tasks-template.md

@@ -10,7 +10,6 @@
    → Extract: tech stack, libraries, structure
 2. Load optional design documents:
    → data-model.md: Extract entities → model tasks
-   → contracts/: Each file → contract test task
    → research.md: Extract decisions → setup tasks
 3. Generate tasks by category:
    → Setup: project init, dependencies, linting
@@ -49,38 +48,37 @@
 
 ## Phase 3.2: Tests First (TDD) ⚠️ MUST COMPLETE BEFORE 3.3
 **CRITICAL: These tests MUST be written and MUST FAIL before ANY implementation**
-- [ ] T004 [P] Contract test for new functionality using ZIO Test framework
-- [ ] T005 [P] Property-based tests for complex logic using ZIO Test Gen
-- [ ] T006 [P] Integration tests for external interactions (RPC, storage)
-- [ ] T007 [P] Performance benchmark tests for critical paths
+- [ ] T004 [P] Property-based tests for complex logic using ZIO Test Gen
+- [ ] T005 [P] Integration tests for external interactions (RPC, storage)
+- [ ] T006 [P] Performance benchmark tests for critical paths
 
 ## Phase 3.3: Core Implementation (ONLY after tests are failing)
-- [ ] T008 [P] Immutable data models with proper type safety
-- [ ] T009 [P] Service layer using ZIO effects and explicit error types
-- [ ] T010 [P] Extend existing abstractions (StateMachine, RPC, LogStore) without breaking changes
-- [ ] T011 Implement functionality using ZIO ecosystem (ZStream, ZLayer, etc.)
-- [ ] T012 Add explicit error handling for all external interactions
-- [ ] T013 Input validation using ZIO Prelude or similar type-safe validation
-- [ ] T014 Resource management using ZIO Scope patterns
+- [ ] T007 [P] Immutable data models with proper type safety
+- [ ] T008 [P] Service layer using ZIO effects and explicit error types
+- [ ] T009 [P] Extend existing abstractions (StateMachine, RPC, LogStore) without breaking changes
+- [ ] T010 Implement functionality using ZIO ecosystem (ZStream, ZLayer, etc.)
+- [ ] T011 Add explicit error handling for all external interactions
+- [ ] T012 Input validation using ZIO Prelude or similar type-safe validation
+- [ ] T013 Resource management using ZIO Scope patterns
 
 ## Phase 3.4: Integration
-- [ ] T015 Integrate with existing storage abstractions using ZIO patterns
-- [ ] T016 Add distributed system error handling (timeouts, retries)
-- [ ] T017 Implement observability using ZIO logging and metrics
-- [ ] T018 Ensure backward compatibility and performance preservation
+- [ ] T014 Integrate with existing storage abstractions using ZIO patterns
+- [ ] T015 Add distributed system error handling (timeouts, retries)
+- [ ] T016 Implement observability using ZIO logging and metrics
+- [ ] T017 Ensure backward compatibility and performance preservation
 
 ## Phase 3.5: Polish
-- [ ] T019 [P] Unit tests covering edge cases and error scenarios
-- [ ] T020 Performance validation and benchmark comparison with baseline
-- [ ] T021 [P] Update documentation and API specifications
-- [ ] T022 Code review for ZIO ecosystem consistency and functional purity
-- [ ] T023 Constitution compliance verification checklist
+- [ ] T018 [P] Unit tests covering edge cases and error scenarios
+- [ ] T019 Performance validation and benchmark comparison with baseline
+- [ ] T020 [P] Update documentation and API specifications
+- [ ] T021 Code review for ZIO ecosystem consistency and functional purity
+- [ ] T022 Constitution compliance verification checklist
 
 ## Dependencies
-- Tests (T004-T007) before implementation (T008-T014)
-- T008 blocks T009, T015
-- T016 blocks T018
-- Implementation before polish (T019-T023)
+- Tests (T004-T006) before implementation (T008-T014)
+- T007 blocks T008, T014
+- T015 blocks T017
+- Implementation before polish (T018-T022)
 
 ## Parallel Example
 ```

client-server-client/src/main/scala/zio/raft/client/PendingRequests.scala

@@ -2,6 +2,7 @@ package zio.raft.client
 
 import zio.*
 import zio.raft.protocol.*
+import scala.math.Ordering
 import scodec.bits.ByteVector
 import java.time.Instant
 
@@ -10,6 +11,10 @@ import java.time.Instant
 case class PendingRequests(
   requests: Map[RequestId, PendingRequests.PendingRequestData]
 ) {
+  def contains(requestId: RequestId): Boolean = requests.contains(requestId)
+  def lowestPendingRequestIdOr(default: RequestId): RequestId =
+    if (requests.isEmpty) default else requests.keys.min(using Ordering.by[RequestId, Long](_.value))
+
   def add(
     requestId: RequestId,
     payload: ByteVector,
@@ -33,7 +38,8 @@ case class PendingRequests(
     ZIO.foldLeft(requests.toList)(this) { case (pending, (requestId, data)) =>
       for {
         now <- Clock.instant
-        request = ClientRequest(requestId, data.payload, now)
+        lowestPendingRequestId = pending.lowestPendingRequestIdOr(requestId)
+        request = ClientRequest(requestId, lowestPendingRequestId, data.payload, now)
         _ <- transport.sendMessage(request).orDie
         _ <- ZIO.logDebug(s"Resending pending request: $requestId")
         updatedData = data.copy(lastSentAt = now)
@@ -47,7 +53,8 @@ case class PendingRequests(
     ZIO.foldLeft(requests.toList)(this) { case (pending, (requestId, data)) =>
       val elapsed = Duration.fromInterval(data.lastSentAt, currentTime)
       if (elapsed > timeout) {
-        val request = ClientRequest(requestId, data.payload, currentTime)
+        val lowestPendingRequestId = pending.lowestPendingRequestIdOr(requestId)
+        val request = ClientRequest(requestId, lowestPendingRequestId, data.payload, currentTime)
         for {
           _ <- transport.sendMessage(request).orDie
           _ <- ZIO.logDebug(s"Resending timed out request: $requestId")
@@ -58,6 +65,12 @@ case class PendingRequests(
       }
     }
   }
+
+  def die(requestId: RequestId, error: Throwable): UIO[PendingRequests] =
+    requests.get(requestId) match {
+      case Some(data) => data.promise.die(error).ignore.as(copy(requests = requests.removed(requestId)))
+      case None       => ZIO.succeed(this)
+    }
 }
 
 object PendingRequests {

client-server-client/src/main/scala/zio/raft/client/RaftClient.scala

@@ -268,6 +268,9 @@ object RaftClient {
           case StreamEvent.ServerMsg(ClientResponse(_, _)) =>
             ZIO.logWarning("Received ClientResponse while connecting, ignoring").as(this)
 
+          case StreamEvent.ServerMsg(_: RequestError) =>
+            ZIO.logWarning("Received RequestError while connecting, ignoring").as(this)
+
           case StreamEvent.ServerMsg(KeepAliveResponse(_)) =>
             ZIO.succeed(this)
 
@@ -405,6 +408,9 @@ object RaftClient {
           case StreamEvent.ServerMsg(KeepAliveResponse(_)) =>
             ZIO.succeed(this)
 
+          case StreamEvent.ServerMsg(_: RequestError) =>
+            ZIO.logWarning("Received RequestError while connecting existing session, ignoring").as(this)
+
           case StreamEvent.ServerMsg(_: ServerRequest) =>
             ZIO.logWarning("Received ServerRequest while connecting, ignoring").as(this)
 
@@ -490,14 +496,23 @@ object RaftClient {
             for {
               requestId <- nextRequestId.next
               now <- Clock.instant
-              request = ClientRequest(requestId, payload, now)
+              lowestPendingRequestId = pendingRequests.lowestPendingRequestIdOr(requestId)
+              request = ClientRequest(requestId, lowestPendingRequestId, payload, now)
               _ <- transport.sendMessage(request).orDie
               newPending = pendingRequests.add(requestId, payload, promise, now)
             } yield copy(pendingRequests = newPending)
 
           case StreamEvent.ServerMsg(ClientResponse(requestId, result)) =>
             pendingRequests.complete(requestId, result).map(newPending => copy(pendingRequests = newPending))
 
+          case StreamEvent.ServerMsg(RequestError(requestId, RequestErrorReason.ResponseEvicted)) =>
+            if (pendingRequests.contains(requestId))
+              ZIO.logError(s"RequestError: ResponseEvicted for request $requestId, terminating client") *>
+                pendingRequests.die(requestId, new RuntimeException("ResponseEvicted")) *>
+                ZIO.dieMessage("ResponseEvicted")
+            else
+              ZIO.logWarning(s"RequestError for non-pending request $requestId, ignoring").as(this)
+
           case StreamEvent.ServerMsg(KeepAliveResponse(_)) =>
             ZIO.succeed(this)
 

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

@@ -0,0 +1,92 @@
+package zio.raft.client
+
+import _root_.zio.*
+import _root_.zio.test.*
+import _root_.zio.test.Assertion.*
+import _root_.zio.raft.protocol.*
+import scodec.bits.ByteVector
+import java.time.Instant
+
+object PendingRequestsSpec extends ZIOSpecDefault {
+
+  private class FakeTransport(ref: Ref[List[ClientMessage]]) extends ClientTransport {
+    override def connect(address: String): ZIO[Any, Throwable, Unit] = ZIO.unit
+    override def disconnect(): ZIO[Any, Throwable, Unit] = ZIO.unit
+    override def sendMessage(message: ClientMessage): ZIO[Any, Throwable, Unit] = ref.update(message :: _).unit
+    override def incomingMessages: _root_.zio.stream.ZStream[Any, Throwable, ServerMessage] =
+      _root_.zio.stream.ZStream.empty
+  }
+
+  def spec = suiteAll("PendingRequests") {
+
+    test("lowestPendingRequestIdOr returns min or default") {
+      val rid1 = RequestId.fromLong(5L)
+      val rid2 = RequestId.fromLong(2L)
+      val rid3 = RequestId.fromLong(9L)
+      for {
+        p <- Promise.make[Throwable, ByteVector]
+        now <- Clock.instant
+        pending = PendingRequests.empty
+          .add(rid1, ByteVector.empty, p, now)
+          .add(rid2, ByteVector.empty, p, now)
+          .add(rid3, ByteVector.empty, p, now)
+      } yield assertTrue(pending.lowestPendingRequestIdOr(rid1) == rid2)
+    }
+
+    test("resendAll includes lowestPendingRequestId = min(pending)") {
+      val rid1 = RequestId.fromLong(1L)
+      val rid2 = RequestId.fromLong(3L)
+      for {
+        sentRef <- Ref.make(List.empty[ClientMessage])
+        transport = new FakeTransport(sentRef)
+        p <- Promise.make[Throwable, ByteVector]
+        now <- Clock.instant
+        pending = PendingRequests.empty
+          .add(rid2, ByteVector.fromValidHex("02"), p, now)
+          .add(rid1, ByteVector.fromValidHex("01"), p, now)
+        _ <- pending.resendAll(transport)
+        sent <- sentRef.get
+        msgs = sent.collect { case m: ClientRequest => m }
+      } yield assertTrue(msgs.nonEmpty) && assertTrue(msgs.forall(_.lowestPendingRequestId == rid1))
+    }
+
+    test("resendExpired includes lowestPendingRequestId = min(pending)") {
+      val rid1 = RequestId.fromLong(2L)
+      val rid2 = RequestId.fromLong(4L)
+      for {
+        sentRef <- Ref.make(List.empty[ClientMessage])
+        transport = new FakeTransport(sentRef)
+        p <- Promise.make[Throwable, ByteVector]
+        createdAt = Instant.parse("2023-01-01T00:00:00Z")
+        pending = PendingRequests(Map(
+          rid1 -> PendingRequests.PendingRequestData(ByteVector.fromValidHex("aa"), p, createdAt, createdAt),
+          rid2 -> PendingRequests.PendingRequestData(ByteVector.fromValidHex("bb"), p, createdAt, createdAt)
+        ))
+        current <- ZIO.succeed(Instant.parse("2023-01-01T00:10:00Z"))
+        _ <- pending.resendExpired(transport, current, 1.minute)
+        sent <- sentRef.get
+        msgs = sent.collect { case m: ClientRequest => m }
+      } yield assertTrue(msgs.nonEmpty) && assertTrue(msgs.forall(_.lowestPendingRequestId == rid1))
+    }
+
+    test("die removes request and completes promise with death") {
+      val rid = RequestId.fromLong(7L)
+      for {
+        p <- Promise.make[Throwable, ByteVector]
+        now <- Clock.instant
+        pending0 = PendingRequests.empty.add(rid, ByteVector.fromValidHex("aa"), p, now)
+        pending1 <- pending0.die(rid, new RuntimeException("boom"))
+        fiber <- p.await.fork
+        exit <- fiber.await
+      } yield assertTrue(!pending1.contains(rid)) && assertTrue(exit.isFailure)
+    }
+
+    test("ignore non-pending die invocation leaves state unchanged") {
+      val rid = RequestId.fromLong(7L)
+      for {
+        pending <- ZIO.succeed(PendingRequests.empty)
+        pending1 <- pending.die(rid, new RuntimeException("boom"))
+      } yield assertTrue(pending1 == pending)
+    }
+  }
+}