docs: add final remaining jetstream parity plan
This commit is contained in:
Joseph Doherty
844
docs/plans/2026-02-23-jetstream-final-remaining-plan.md
Normal file
844
docs/plans/2026-02-23-jetstream-final-remaining-plan.md
Normal file
@@ -0,0 +1,844 @@
|
|||||||
|
# JetStream Final Remaining Parity Implementation Plan
|
||||||
|
|
||||||
|
> **For Codex:** REQUIRED SUB-SKILL: Use `executeplan` to implement this plan task-by-task.
|
||||||
|
|
||||||
|
**Goal:** Close all remaining JetStream parity gaps (and required transport prerequisites) between Go and .NET so JetStream entries are no longer marked partial in `differences.md` except explicitly documented external blockers.
|
||||||
|
|
||||||
|
**Architecture:** Implement prerequisites first (route/gateway/leaf wire behavior), then complete JetStream API and runtime semantics on top of real inter-server transport, and finally harden storage/RAFT and monitoring evidence. Use parity-map-driven development: every Go feature gap must map to concrete .NET code and test proof.
|
||||||
|
|
||||||
|
**Tech Stack:** .NET 10, C# 14, xUnit 3, Shouldly, NSubstitute, bash tooling, ripgrep, System.Text.Json.
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
**Execution guardrails**
|
||||||
|
- Use `@test-driven-development` in every task.
|
||||||
|
- If behavior diverges from expected protocol semantics, switch to `@systematic-debugging` before modifying production code.
|
||||||
|
- Use a dedicated worktree for execution.
|
||||||
|
- Before completion claims, run `@verification-before-completion` commands.
|
||||||
|
|
||||||
|
### Task 1: Regenerate and Enforce Go-vs-.NET JetStream Subject Gap Inventory
|
||||||
|
|
||||||
|
**Files:**
|
||||||
|
- Modify: `scripts/jetstream/extract-go-js-api.sh`
|
||||||
|
- Modify: `docs/plans/2026-02-23-jetstream-remaining-parity-map.md`
|
||||||
|
- Create: `tests/NATS.Server.Tests/JetStreamApiGapInventoryTests.cs`
|
||||||
|
|
||||||
|
**Step 1: Write the failing test**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
[Fact]
|
||||||
|
public void Parity_map_has_no_unclassified_go_js_api_subjects()
|
||||||
|
{
|
||||||
|
var gap = JetStreamApiGapInventory.Load();
|
||||||
|
gap.UnclassifiedSubjects.ShouldBeEmpty();
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 2: Run test to verify it fails**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamApiGapInventoryTests.Parity_map_has_no_unclassified_go_js_api_subjects" -v minimal`
|
||||||
|
Expected: FAIL with listed missing subjects (`SERVER.REMOVE`, `ACCOUNT.PURGE`, `STREAM.PEER.REMOVE`, etc.).
|
||||||
|
|
||||||
|
**Step 3: Write minimal implementation**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
#!/usr/bin/env bash
|
||||||
|
set -euo pipefail
|
||||||
|
perl -nle 'while(/"(\$JS\.API[^"]+)"/g){print $1}' golang/nats-server/server/jetstream_api.go | sort -u
|
||||||
|
```
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
public static JetStreamApiGapInventory Load()
|
||||||
|
{
|
||||||
|
// compare extracted Go subject set with mapped .NET subject handlers
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 4: Run test to verify it passes**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamApiGapInventoryTests" -v minimal`
|
||||||
|
Expected: PASS.
|
||||||
|
|
||||||
|
**Step 5: Commit**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git add scripts/jetstream/extract-go-js-api.sh docs/plans/2026-02-23-jetstream-remaining-parity-map.md tests/NATS.Server.Tests/JetStreamApiGapInventoryTests.cs
|
||||||
|
git commit -m "test: enforce jetstream api gap inventory parity map"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Task 2: Enforce Multi-Client-Type Command Routing and Inter-Server Opcodes
|
||||||
|
|
||||||
|
**Files:**
|
||||||
|
- Modify: `src/NATS.Server/Protocol/NatsParser.cs`
|
||||||
|
- Modify: `src/NATS.Server/Protocol/ClientCommandMatrix.cs`
|
||||||
|
- Modify: `src/NATS.Server/NatsClient.cs`
|
||||||
|
- Test: `tests/NATS.Server.Tests/ClientKindProtocolRoutingTests.cs`
|
||||||
|
|
||||||
|
**Step 1: Write the failing test**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
[Fact]
|
||||||
|
public void Client_kind_rejects_RSplus_for_non_route_connection()
|
||||||
|
{
|
||||||
|
var matrix = new ClientCommandMatrix();
|
||||||
|
matrix.IsAllowed(ClientKind.Client, "RS+").ShouldBeFalse();
|
||||||
|
matrix.IsAllowed(ClientKind.Router, "RS+").ShouldBeTrue();
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 2: Run test to verify it fails**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~ClientKindProtocolRoutingTests" -v minimal`
|
||||||
|
Expected: FAIL for missing/incorrect kind restrictions on RS+/RS-/RMSG/A+/A-/LS+/LS-/LMSG.
|
||||||
|
|
||||||
|
**Step 3: Write minimal implementation**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
(ClientKind.Router, "RS+") => true,
|
||||||
|
(ClientKind.Router, "RS-") => true,
|
||||||
|
(ClientKind.Router, "RMSG") => true,
|
||||||
|
(ClientKind.Leaf, "LS+") => true,
|
||||||
|
(ClientKind.Leaf, "LS-") => true,
|
||||||
|
(ClientKind.Leaf, "LMSG") => true,
|
||||||
|
_ => false,
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 4: Run test to verify it passes**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~ClientKindProtocolRoutingTests" -v minimal`
|
||||||
|
Expected: PASS.
|
||||||
|
|
||||||
|
**Step 5: Commit**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git add src/NATS.Server/Protocol/NatsParser.cs src/NATS.Server/Protocol/ClientCommandMatrix.cs src/NATS.Server/NatsClient.cs tests/NATS.Server.Tests/ClientKindProtocolRoutingTests.cs
|
||||||
|
git commit -m "feat: enforce client-kind protocol routing for inter-server ops"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Task 3: Implement Route Wire RS+/RS- Subscription Propagation
|
||||||
|
|
||||||
|
**Files:**
|
||||||
|
- Modify: `src/NATS.Server/Routes/RouteConnection.cs`
|
||||||
|
- Modify: `src/NATS.Server/Routes/RouteManager.cs`
|
||||||
|
- Modify: `src/NATS.Server/Subscriptions/SubList.cs`
|
||||||
|
- Test: `tests/NATS.Server.Tests/RouteWireSubscriptionProtocolTests.cs`
|
||||||
|
|
||||||
|
**Step 1: Write the failing test**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
[Fact]
|
||||||
|
public async Task RSplus_RSminus_frames_propagate_remote_interest_over_socket()
|
||||||
|
{
|
||||||
|
await using var fx = await RouteFixture.StartTwoNodeClusterAsync();
|
||||||
|
await fx.SendRouteSubFrameAsync("foo.*");
|
||||||
|
(await fx.ServerAHasRemoteInterestAsync("foo.bar")).ShouldBeTrue();
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 2: Run test to verify it fails**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~RouteWireSubscriptionProtocolTests" -v minimal`
|
||||||
|
Expected: FAIL because propagation is currently in-process and not RS+/RS- wire-driven.
|
||||||
|
|
||||||
|
**Step 3: Write minimal implementation**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
await WriteOpAsync($"RS+ {subject}");
|
||||||
|
```
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
if (op == "RS+") _remoteSubSink(new RemoteSubscription(subject, queue, remoteServerId));
|
||||||
|
if (op == "RS-") _remoteSubSink(RemoteSubscription.Removal(subject, queue, remoteServerId));
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 4: Run test to verify it passes**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~RouteWireSubscriptionProtocolTests" -v minimal`
|
||||||
|
Expected: PASS.
|
||||||
|
|
||||||
|
**Step 5: Commit**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git add src/NATS.Server/Routes/RouteConnection.cs src/NATS.Server/Routes/RouteManager.cs src/NATS.Server/Subscriptions/SubList.cs tests/NATS.Server.Tests/RouteWireSubscriptionProtocolTests.cs
|
||||||
|
git commit -m "feat: implement route RS+ RS- wire subscription protocol"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Task 4: Implement Route RMSG Forwarding to Remote Subscribers
|
||||||
|
|
||||||
|
**Files:**
|
||||||
|
- Modify: `src/NATS.Server/Routes/RouteConnection.cs`
|
||||||
|
- Modify: `src/NATS.Server/Routes/RouteManager.cs`
|
||||||
|
- Modify: `src/NATS.Server/NatsServer.cs`
|
||||||
|
- Test: `tests/NATS.Server.Tests/RouteRmsgForwardingTests.cs`
|
||||||
|
|
||||||
|
**Step 1: Write the failing test**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
[Fact]
|
||||||
|
public async Task Publish_on_serverA_reaches_remote_subscriber_on_serverB_via_RMSG()
|
||||||
|
{
|
||||||
|
await using var fx = await RouteFixture.StartTwoNodeClusterAsync();
|
||||||
|
await fx.SubscribeOnServerBAsync("foo.>");
|
||||||
|
|
||||||
|
await fx.PublishFromServerAAsync("foo.bar", "payload");
|
||||||
|
(await fx.ReadServerBMessageAsync()).ShouldContain("payload");
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 2: Run test to verify it fails**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~RouteRmsgForwardingTests" -v minimal`
|
||||||
|
Expected: FAIL because remote messages are not forwarded.
|
||||||
|
|
||||||
|
**Step 3: Write minimal implementation**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
if (hasRemoteInterest)
|
||||||
|
await route.SendRmsgAsync(subject, reply, headers, payload, ct);
|
||||||
|
```
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
if (op == "RMSG") _server.ProcessRoutedMessage(parsed);
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 4: Run test to verify it passes**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~RouteRmsgForwardingTests" -v minimal`
|
||||||
|
Expected: PASS.
|
||||||
|
|
||||||
|
**Step 5: Commit**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git add src/NATS.Server/Routes/RouteConnection.cs src/NATS.Server/Routes/RouteManager.cs src/NATS.Server/NatsServer.cs tests/NATS.Server.Tests/RouteRmsgForwardingTests.cs
|
||||||
|
git commit -m "feat: forward remote messages over route RMSG"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Task 5: Add Route Pooling Baseline (3 Connections per Peer)
|
||||||
|
|
||||||
|
**Files:**
|
||||||
|
- Modify: `src/NATS.Server/Routes/RouteManager.cs`
|
||||||
|
- Modify: `src/NATS.Server/Configuration/ClusterOptions.cs`
|
||||||
|
- Test: `tests/NATS.Server.Tests/RoutePoolTests.cs`
|
||||||
|
|
||||||
|
**Step 1: Write the failing test**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
[Fact]
|
||||||
|
public async Task Route_manager_establishes_default_pool_of_three_links_per_peer()
|
||||||
|
{
|
||||||
|
await using var fx = await RouteFixture.StartTwoNodeClusterAsync();
|
||||||
|
(await fx.ServerARouteLinkCountToServerBAsync()).ShouldBe(3);
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 2: Run test to verify it fails**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~RoutePoolTests" -v minimal`
|
||||||
|
Expected: FAIL because one connection per peer is used.
|
||||||
|
|
||||||
|
**Step 3: Write minimal implementation**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
public int PoolSize { get; set; } = 3;
|
||||||
|
for (var i = 0; i < _options.PoolSize; i++)
|
||||||
|
_ = ConnectToRouteWithRetryAsync(route, ct);
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 4: Run test to verify it passes**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~RoutePoolTests" -v minimal`
|
||||||
|
Expected: PASS.
|
||||||
|
|
||||||
|
**Step 5: Commit**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git add src/NATS.Server/Routes/RouteManager.cs src/NATS.Server/Configuration/ClusterOptions.cs tests/NATS.Server.Tests/RoutePoolTests.cs
|
||||||
|
git commit -m "feat: add route connection pooling baseline"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Task 6: Replace Gateway Stub with Functional Handshake and Forwarding Baseline
|
||||||
|
|
||||||
|
**Files:**
|
||||||
|
- Modify: `src/NATS.Server/Gateways/GatewayManager.cs`
|
||||||
|
- Modify: `src/NATS.Server/Gateways/GatewayConnection.cs`
|
||||||
|
- Modify: `src/NATS.Server/NatsServer.cs`
|
||||||
|
- Test: `tests/NATS.Server.Tests/GatewayProtocolTests.cs`
|
||||||
|
|
||||||
|
**Step 1: Write the failing test**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
[Fact]
|
||||||
|
public async Task Gateway_link_establishes_and_forwards_interested_message()
|
||||||
|
{
|
||||||
|
await using var fx = await GatewayFixture.StartTwoClustersAsync();
|
||||||
|
await fx.SubscribeRemoteClusterAsync("g.>");
|
||||||
|
await fx.PublishLocalClusterAsync("g.test", "hello");
|
||||||
|
(await fx.ReadRemoteClusterMessageAsync()).ShouldContain("hello");
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 2: Run test to verify it fails**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~GatewayProtocolTests" -v minimal`
|
||||||
|
Expected: FAIL due to gateway no-op manager.
|
||||||
|
|
||||||
|
**Step 3: Write minimal implementation**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
public Task StartAsync(CancellationToken ct)
|
||||||
|
{
|
||||||
|
// listen/connect handshake and track gateway links
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 4: Run test to verify it passes**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~GatewayProtocolTests" -v minimal`
|
||||||
|
Expected: PASS.
|
||||||
|
|
||||||
|
**Step 5: Commit**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git add src/NATS.Server/Gateways/GatewayManager.cs src/NATS.Server/Gateways/GatewayConnection.cs src/NATS.Server/NatsServer.cs tests/NATS.Server.Tests/GatewayProtocolTests.cs
|
||||||
|
git commit -m "feat: replace gateway stub with functional protocol baseline"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Task 7: Replace Leaf Stub with Functional LS+/LS-/LMSG Baseline
|
||||||
|
|
||||||
|
**Files:**
|
||||||
|
- Modify: `src/NATS.Server/LeafNodes/LeafNodeManager.cs`
|
||||||
|
- Modify: `src/NATS.Server/LeafNodes/LeafConnection.cs`
|
||||||
|
- Modify: `src/NATS.Server/NatsServer.cs`
|
||||||
|
- Test: `tests/NATS.Server.Tests/LeafProtocolTests.cs`
|
||||||
|
|
||||||
|
**Step 1: Write the failing test**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
[Fact]
|
||||||
|
public async Task Leaf_link_propagates_subscription_and_message_flow()
|
||||||
|
{
|
||||||
|
await using var fx = await LeafFixture.StartHubSpokeAsync();
|
||||||
|
await fx.SubscribeSpokeAsync("leaf.>");
|
||||||
|
await fx.PublishHubAsync("leaf.msg", "x");
|
||||||
|
(await fx.ReadSpokeMessageAsync()).ShouldContain("x");
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 2: Run test to verify it fails**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~LeafProtocolTests" -v minimal`
|
||||||
|
Expected: FAIL due to leaf no-op manager.
|
||||||
|
|
||||||
|
**Step 3: Write minimal implementation**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
if (op == "LS+") ApplyLeafSubscription(...);
|
||||||
|
if (op == "LMSG") ProcessLeafMessage(...);
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 4: Run test to verify it passes**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~LeafProtocolTests" -v minimal`
|
||||||
|
Expected: PASS.
|
||||||
|
|
||||||
|
**Step 5: Commit**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git add src/NATS.Server/LeafNodes/LeafNodeManager.cs src/NATS.Server/LeafNodes/LeafConnection.cs src/NATS.Server/NatsServer.cs tests/NATS.Server.Tests/LeafProtocolTests.cs
|
||||||
|
git commit -m "feat: replace leaf stub with functional protocol baseline"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Task 8: Add Missing JetStream Control APIs (`SERVER.REMOVE`, `ACCOUNT.PURGE`, Move/Cancel Move)
|
||||||
|
|
||||||
|
**Files:**
|
||||||
|
- Modify: `src/NATS.Server/JetStream/Api/JetStreamApiSubjects.cs`
|
||||||
|
- Modify: `src/NATS.Server/JetStream/Api/JetStreamApiRouter.cs`
|
||||||
|
- Create: `src/NATS.Server/JetStream/Api/Handlers/AccountControlApiHandlers.cs`
|
||||||
|
- Test: `tests/NATS.Server.Tests/JetStreamAccountControlApiTests.cs`
|
||||||
|
|
||||||
|
**Step 1: Write the failing test**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
[Fact]
|
||||||
|
public void Account_and_server_control_subjects_are_routable()
|
||||||
|
{
|
||||||
|
var r = new JetStreamApiRouter(new StreamManager(), new ConsumerManager());
|
||||||
|
r.Route("$JS.API.SERVER.REMOVE", "{}"u8).Error.ShouldBeNull();
|
||||||
|
r.Route("$JS.API.ACCOUNT.PURGE.ACC", "{}"u8).Error.ShouldBeNull();
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 2: Run test to verify it fails**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamAccountControlApiTests" -v minimal`
|
||||||
|
Expected: FAIL with NotFound responses.
|
||||||
|
|
||||||
|
**Step 3: Write minimal implementation**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
public const string ServerRemove = "$JS.API.SERVER.REMOVE";
|
||||||
|
public const string AccountPurge = "$JS.API.ACCOUNT.PURGE.";
|
||||||
|
```
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
if (subject.Equals(JetStreamApiSubjects.ServerRemove, StringComparison.Ordinal))
|
||||||
|
return AccountControlApiHandlers.HandleServerRemove();
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 4: Run test to verify it passes**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamAccountControlApiTests" -v minimal`
|
||||||
|
Expected: PASS.
|
||||||
|
|
||||||
|
**Step 5: Commit**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git add src/NATS.Server/JetStream/Api/JetStreamApiSubjects.cs src/NATS.Server/JetStream/Api/JetStreamApiRouter.cs src/NATS.Server/JetStream/Api/Handlers/AccountControlApiHandlers.cs tests/NATS.Server.Tests/JetStreamAccountControlApiTests.cs
|
||||||
|
git commit -m "feat: add missing jetstream account and server control apis"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Task 9: Add Missing Cluster JetStream APIs (`STREAM.PEER.REMOVE`, `CONSUMER.LEADER.STEPDOWN`)
|
||||||
|
|
||||||
|
**Files:**
|
||||||
|
- Modify: `src/NATS.Server/JetStream/Api/JetStreamApiSubjects.cs`
|
||||||
|
- Modify: `src/NATS.Server/JetStream/Api/Handlers/ClusterControlApiHandlers.cs`
|
||||||
|
- Modify: `src/NATS.Server/JetStream/Api/JetStreamApiRouter.cs`
|
||||||
|
- Test: `tests/NATS.Server.Tests/JetStreamClusterControlExtendedApiTests.cs`
|
||||||
|
|
||||||
|
**Step 1: Write the failing test**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
[Fact]
|
||||||
|
public async Task Peer_remove_and_consumer_stepdown_subjects_return_success_shape()
|
||||||
|
{
|
||||||
|
await using var fx = await JetStreamClusterFixture.StartAsync(nodes: 3);
|
||||||
|
(await fx.RequestAsync("$JS.API.STREAM.PEER.REMOVE.ORDERS", "{\"peer\":\"n2\"}")).Success.ShouldBeTrue();
|
||||||
|
(await fx.RequestAsync("$JS.API.CONSUMER.LEADER.STEPDOWN.ORDERS.DUR", "{}")).Success.ShouldBeTrue();
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 2: Run test to verify it fails**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamClusterControlExtendedApiTests" -v minimal`
|
||||||
|
Expected: FAIL because these routes are missing.
|
||||||
|
|
||||||
|
**Step 3: Write minimal implementation**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
public const string StreamPeerRemove = "$JS.API.STREAM.PEER.REMOVE.";
|
||||||
|
public const string ConsumerLeaderStepdown = "$JS.API.CONSUMER.LEADER.STEPDOWN.";
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 4: Run test to verify it passes**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamClusterControlExtendedApiTests" -v minimal`
|
||||||
|
Expected: PASS.
|
||||||
|
|
||||||
|
**Step 5: Commit**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git add src/NATS.Server/JetStream/Api/JetStreamApiSubjects.cs src/NATS.Server/JetStream/Api/Handlers/ClusterControlApiHandlers.cs src/NATS.Server/JetStream/Api/JetStreamApiRouter.cs tests/NATS.Server.Tests/JetStreamClusterControlExtendedApiTests.cs
|
||||||
|
git commit -m "feat: add extended jetstream cluster control apis"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Task 10: Implement Stream Policy Runtime Semantics (`MaxBytes`, `MaxAge`, `MaxMsgsPer`, `DiscardNew`)
|
||||||
|
|
||||||
|
**Files:**
|
||||||
|
- Modify: `src/NATS.Server/JetStream/Models/StreamConfig.cs`
|
||||||
|
- Modify: `src/NATS.Server/JetStream/StreamManager.cs`
|
||||||
|
- Modify: `src/NATS.Server/JetStream/Storage/IStreamStore.cs`
|
||||||
|
- Modify: `src/NATS.Server/JetStream/Storage/MemStore.cs`
|
||||||
|
- Modify: `src/NATS.Server/JetStream/Storage/FileStore.cs`
|
||||||
|
- Test: `tests/NATS.Server.Tests/JetStreamStreamPolicyRuntimeTests.cs`
|
||||||
|
|
||||||
|
**Step 1: Write the failing test**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
[Fact]
|
||||||
|
public async Task Discard_new_rejects_publish_when_max_bytes_exceeded()
|
||||||
|
{
|
||||||
|
await using var fx = await JetStreamApiFixture.StartWithStreamConfigAsync(new StreamConfig { Name = "S", Subjects = ["s.*"], MaxBytes = 2, Discard = DiscardPolicy.New });
|
||||||
|
(await fx.PublishAndGetAckAsync("s.a", "12")).ErrorCode.ShouldBeNull();
|
||||||
|
(await fx.PublishAndGetAckAsync("s.a", "34")).ErrorCode.ShouldNotBeNull();
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 2: Run test to verify it fails**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamStreamPolicyRuntimeTests" -v minimal`
|
||||||
|
Expected: FAIL because runtime enforces only `MaxMsgs`.
|
||||||
|
|
||||||
|
**Step 3: Write minimal implementation**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
if (cfg.MaxBytes > 0 && state.Bytes + payload.Length > cfg.MaxBytes && cfg.Discard == DiscardPolicy.New)
|
||||||
|
return PublishDecision.Reject(10054, "maximum bytes exceeded");
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 4: Run test to verify it passes**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamStreamPolicyRuntimeTests" -v minimal`
|
||||||
|
Expected: PASS.
|
||||||
|
|
||||||
|
**Step 5: Commit**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git add src/NATS.Server/JetStream/Models/StreamConfig.cs src/NATS.Server/JetStream/StreamManager.cs src/NATS.Server/JetStream/Storage/IStreamStore.cs src/NATS.Server/JetStream/Storage/MemStore.cs src/NATS.Server/JetStream/Storage/FileStore.cs tests/NATS.Server.Tests/JetStreamStreamPolicyRuntimeTests.cs
|
||||||
|
git commit -m "feat: enforce stream runtime policies maxbytes maxage maxmsgsper discard"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Task 11: Implement Storage Type Selection and Config Mapping for JetStream Streams
|
||||||
|
|
||||||
|
**Files:**
|
||||||
|
- Modify: `src/NATS.Server/JetStream/Models/StreamConfig.cs`
|
||||||
|
- Modify: `src/NATS.Server/JetStream/StreamManager.cs`
|
||||||
|
- Modify: `src/NATS.Server/Configuration/ConfigProcessor.cs`
|
||||||
|
- Test: `tests/NATS.Server.Tests/JetStreamStorageSelectionTests.cs`
|
||||||
|
|
||||||
|
**Step 1: Write the failing test**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
[Fact]
|
||||||
|
public async Task Stream_with_storage_file_uses_filestore_backend()
|
||||||
|
{
|
||||||
|
await using var fx = await JetStreamApiFixture.StartWithStreamJsonAsync("{\"name\":\"S\",\"subjects\":[\"s.*\"],\"storage\":\"file\"}");
|
||||||
|
(await fx.GetStreamBackendTypeAsync("S")).ShouldBe("file");
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 2: Run test to verify it fails**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamStorageSelectionTests" -v minimal`
|
||||||
|
Expected: FAIL because memstore is always used.
|
||||||
|
|
||||||
|
**Step 3: Write minimal implementation**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
var store = config.Storage switch
|
||||||
|
{
|
||||||
|
StorageType.File => new FileStore(new FileStoreOptions { Directory = ResolveStoreDir(config.Name) }),
|
||||||
|
_ => new MemStore(),
|
||||||
|
};
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 4: Run test to verify it passes**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamStorageSelectionTests" -v minimal`
|
||||||
|
Expected: PASS.
|
||||||
|
|
||||||
|
**Step 5: Commit**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git add src/NATS.Server/JetStream/Models/StreamConfig.cs src/NATS.Server/JetStream/StreamManager.cs src/NATS.Server/Configuration/ConfigProcessor.cs tests/NATS.Server.Tests/JetStreamStorageSelectionTests.cs
|
||||||
|
git commit -m "feat: select jetstream storage backend per stream config"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Task 12: Implement Consumer Completeness (`Ephemeral`, `FilterSubjects`, `MaxAckPending`, `DeliverPolicy`)
|
||||||
|
|
||||||
|
**Files:**
|
||||||
|
- Modify: `src/NATS.Server/JetStream/Models/ConsumerConfig.cs`
|
||||||
|
- Modify: `src/NATS.Server/JetStream/ConsumerManager.cs`
|
||||||
|
- Modify: `src/NATS.Server/JetStream/Consumers/PullConsumerEngine.cs`
|
||||||
|
- Modify: `src/NATS.Server/JetStream/Consumers/PushConsumerEngine.cs`
|
||||||
|
- Test: `tests/NATS.Server.Tests/JetStreamConsumerSemanticsTests.cs`
|
||||||
|
|
||||||
|
**Step 1: Write the failing test**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
[Fact]
|
||||||
|
public async Task Consumer_with_filter_subjects_only_receives_matching_messages()
|
||||||
|
{
|
||||||
|
await using var fx = await JetStreamApiFixture.StartWithMultiFilterConsumerAsync();
|
||||||
|
await fx.PublishAndGetAckAsync("orders.created", "1");
|
||||||
|
await fx.PublishAndGetAckAsync("payments.settled", "2");
|
||||||
|
|
||||||
|
var batch = await fx.FetchAsync("ORDERS", "CF", 10);
|
||||||
|
batch.Messages.All(m => m.Subject.StartsWith("orders.", StringComparison.Ordinal)).ShouldBeTrue();
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 2: Run test to verify it fails**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamConsumerSemanticsTests" -v minimal`
|
||||||
|
Expected: FAIL because only single filter and limited policy semantics exist.
|
||||||
|
|
||||||
|
**Step 3: Write minimal implementation**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
public List<string> FilterSubjects { get; set; } = [];
|
||||||
|
if (config.FilterSubjects.Count > 0)
|
||||||
|
include = config.FilterSubjects.Any(f => SubjectMatch.MatchLiteral(msg.Subject, f));
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 4: Run test to verify it passes**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamConsumerSemanticsTests" -v minimal`
|
||||||
|
Expected: PASS.
|
||||||
|
|
||||||
|
**Step 5: Commit**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git add src/NATS.Server/JetStream/Models/ConsumerConfig.cs src/NATS.Server/JetStream/ConsumerManager.cs src/NATS.Server/JetStream/Consumers/PullConsumerEngine.cs src/NATS.Server/JetStream/Consumers/PushConsumerEngine.cs tests/NATS.Server.Tests/JetStreamConsumerSemanticsTests.cs
|
||||||
|
git commit -m "feat: complete consumer filters and delivery semantics"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Task 13: Implement Replay/Backoff/Flow Control and Rate Limits
|
||||||
|
|
||||||
|
**Files:**
|
||||||
|
- Modify: `src/NATS.Server/JetStream/Consumers/AckProcessor.cs`
|
||||||
|
- Modify: `src/NATS.Server/JetStream/Consumers/PushConsumerEngine.cs`
|
||||||
|
- Modify: `src/NATS.Server/JetStream/Consumers/PullConsumerEngine.cs`
|
||||||
|
- Test: `tests/NATS.Server.Tests/JetStreamFlowReplayBackoffTests.cs`
|
||||||
|
|
||||||
|
**Step 1: Write the failing test**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
[Fact]
|
||||||
|
public async Task Replay_original_respects_message_timestamps_with_backoff_redelivery()
|
||||||
|
{
|
||||||
|
await using var fx = await JetStreamApiFixture.StartWithReplayOriginalConsumerAsync();
|
||||||
|
var sw = Stopwatch.StartNew();
|
||||||
|
await fx.FetchAsync("ORDERS", "RO", 1);
|
||||||
|
sw.ElapsedMilliseconds.ShouldBeGreaterThanOrEqualTo(50);
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 2: Run test to verify it fails**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamFlowReplayBackoffTests" -v minimal`
|
||||||
|
Expected: FAIL because replay/backoff/rate semantics are incomplete.
|
||||||
|
|
||||||
|
**Step 3: Write minimal implementation**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
if (config.ReplayPolicy == ReplayPolicy.Original)
|
||||||
|
await Task.Delay(originalDelay, ct);
|
||||||
|
```
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
var next = backoff[min(deliveryCount, backoff.Length - 1)];
|
||||||
|
_ackProcessor.Register(seq, next);
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 4: Run test to verify it passes**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamFlowReplayBackoffTests" -v minimal`
|
||||||
|
Expected: PASS.
|
||||||
|
|
||||||
|
**Step 5: Commit**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git add src/NATS.Server/JetStream/Consumers/AckProcessor.cs src/NATS.Server/JetStream/Consumers/PushConsumerEngine.cs src/NATS.Server/JetStream/Consumers/PullConsumerEngine.cs tests/NATS.Server.Tests/JetStreamFlowReplayBackoffTests.cs
|
||||||
|
git commit -m "feat: implement replay backoff flow-control and rate behaviors"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Task 14: Complete Mirror/Source Advanced Semantics (`Sources[]`, transforms, cross-account guardrails)
|
||||||
|
|
||||||
|
**Files:**
|
||||||
|
- Modify: `src/NATS.Server/JetStream/Models/StreamConfig.cs`
|
||||||
|
- Modify: `src/NATS.Server/JetStream/MirrorSource/MirrorCoordinator.cs`
|
||||||
|
- Modify: `src/NATS.Server/JetStream/MirrorSource/SourceCoordinator.cs`
|
||||||
|
- Modify: `src/NATS.Server/JetStream/StreamManager.cs`
|
||||||
|
- Test: `tests/NATS.Server.Tests/JetStreamMirrorSourceAdvancedTests.cs`
|
||||||
|
|
||||||
|
**Step 1: Write the failing test**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
[Fact]
|
||||||
|
public async Task Stream_with_multiple_sources_aggregates_messages_in_order()
|
||||||
|
{
|
||||||
|
await using var fx = await JetStreamApiFixture.StartWithMultipleSourcesAsync();
|
||||||
|
await fx.PublishToSourceAsync("SRC1", "a.1", "1");
|
||||||
|
await fx.PublishToSourceAsync("SRC2", "b.1", "2");
|
||||||
|
|
||||||
|
(await fx.GetStreamStateAsync("AGG")).Messages.ShouldBe(2);
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 2: Run test to verify it fails**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamMirrorSourceAdvancedTests" -v minimal`
|
||||||
|
Expected: FAIL because only single `Source` is supported.
|
||||||
|
|
||||||
|
**Step 3: Write minimal implementation**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
public List<StreamSourceConfig> Sources { get; set; } = [];
|
||||||
|
foreach (var source in config.Sources)
|
||||||
|
RegisterSource(source);
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 4: Run test to verify it passes**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamMirrorSourceAdvancedTests" -v minimal`
|
||||||
|
Expected: PASS.
|
||||||
|
|
||||||
|
**Step 5: Commit**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git add src/NATS.Server/JetStream/Models/StreamConfig.cs src/NATS.Server/JetStream/MirrorSource/MirrorCoordinator.cs src/NATS.Server/JetStream/MirrorSource/SourceCoordinator.cs src/NATS.Server/JetStream/StreamManager.cs tests/NATS.Server.Tests/JetStreamMirrorSourceAdvancedTests.cs
|
||||||
|
git commit -m "feat: complete mirror and source advanced semantics"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Task 15: Upgrade RAFT from In-Memory Coordination to Transport/Persistence Baseline
|
||||||
|
|
||||||
|
**Files:**
|
||||||
|
- Modify: `src/NATS.Server/Raft/RaftNode.cs`
|
||||||
|
- Modify: `src/NATS.Server/Raft/RaftReplicator.cs`
|
||||||
|
- Modify: `src/NATS.Server/Raft/RaftLog.cs`
|
||||||
|
- Create: `src/NATS.Server/Raft/RaftTransport.cs`
|
||||||
|
- Test: `tests/NATS.Server.Tests/RaftTransportPersistenceTests.cs`
|
||||||
|
|
||||||
|
**Step 1: Write the failing test**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
[Fact]
|
||||||
|
public async Task Raft_node_recovers_log_and_term_after_restart()
|
||||||
|
{
|
||||||
|
var fx = await RaftFixture.StartPersistentClusterAsync();
|
||||||
|
var idx = await fx.Leader.ProposeAsync("cmd", default);
|
||||||
|
await fx.RestartNodeAsync("n2");
|
||||||
|
(await fx.ReadNodeAppliedIndexAsync("n2")).ShouldBeGreaterThanOrEqualTo(idx);
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 2: Run test to verify it fails**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~RaftTransportPersistenceTests" -v minimal`
|
||||||
|
Expected: FAIL because no persistent raft transport/log baseline exists.
|
||||||
|
|
||||||
|
**Step 3: Write minimal implementation**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
public interface IRaftTransport
|
||||||
|
{
|
||||||
|
Task<IReadOnlyList<AppendResult>> AppendEntriesAsync(...);
|
||||||
|
Task<VoteResponse> RequestVoteAsync(...);
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
public sealed class RaftLog
|
||||||
|
{
|
||||||
|
public Task PersistAsync(string path, CancellationToken ct) { ... }
|
||||||
|
public static Task<RaftLog> LoadAsync(string path, CancellationToken ct) { ... }
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 4: Run test to verify it passes**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~RaftTransportPersistenceTests" -v minimal`
|
||||||
|
Expected: PASS.
|
||||||
|
|
||||||
|
**Step 5: Commit**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git add src/NATS.Server/Raft/RaftNode.cs src/NATS.Server/Raft/RaftReplicator.cs src/NATS.Server/Raft/RaftLog.cs src/NATS.Server/Raft/RaftTransport.cs tests/NATS.Server.Tests/RaftTransportPersistenceTests.cs
|
||||||
|
git commit -m "feat: add raft transport and persistence baseline"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Task 16: Replace Monitoring Stubs (`/routez`, `/gatewayz`, `/leafz`, `/accountz`, `/accstatz`)
|
||||||
|
|
||||||
|
**Files:**
|
||||||
|
- Modify: `src/NATS.Server/Monitoring/MonitorServer.cs`
|
||||||
|
- Create: `src/NATS.Server/Monitoring/RoutezHandler.cs`
|
||||||
|
- Create: `src/NATS.Server/Monitoring/GatewayzHandler.cs`
|
||||||
|
- Create: `src/NATS.Server/Monitoring/LeafzHandler.cs`
|
||||||
|
- Create: `src/NATS.Server/Monitoring/AccountzHandler.cs`
|
||||||
|
- Test: `tests/NATS.Server.Tests/MonitorClusterEndpointTests.cs`
|
||||||
|
|
||||||
|
**Step 1: Write the failing test**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
[Fact]
|
||||||
|
public async Task Routez_gatewayz_leafz_accountz_return_non_stub_runtime_data()
|
||||||
|
{
|
||||||
|
await using var fx = await MonitorFixture.StartClusterEnabledAsync();
|
||||||
|
(await fx.GetJsonAsync("/routez")).ShouldContain("routes");
|
||||||
|
(await fx.GetJsonAsync("/gatewayz")).ShouldContain("gateways");
|
||||||
|
(await fx.GetJsonAsync("/leafz")).ShouldContain("leafs");
|
||||||
|
(await fx.GetJsonAsync("/accountz")).ShouldContain("accounts");
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 2: Run test to verify it fails**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~MonitorClusterEndpointTests" -v minimal`
|
||||||
|
Expected: FAIL because endpoints currently return stubs.
|
||||||
|
|
||||||
|
**Step 3: Write minimal implementation**
|
||||||
|
|
||||||
|
```csharp
|
||||||
|
_app.MapGet(basePath + "/routez", () => _routezHandler.Build());
|
||||||
|
_app.MapGet(basePath + "/gatewayz", () => _gatewayzHandler.Build());
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 4: Run test to verify it passes**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~MonitorClusterEndpointTests" -v minimal`
|
||||||
|
Expected: PASS.
|
||||||
|
|
||||||
|
**Step 5: Commit**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git add src/NATS.Server/Monitoring/MonitorServer.cs src/NATS.Server/Monitoring/RoutezHandler.cs src/NATS.Server/Monitoring/GatewayzHandler.cs src/NATS.Server/Monitoring/LeafzHandler.cs src/NATS.Server/Monitoring/AccountzHandler.cs tests/NATS.Server.Tests/MonitorClusterEndpointTests.cs
|
||||||
|
git commit -m "feat: replace cluster monitoring endpoint stubs"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Task 17: Final Strict Gate, Parity Map Closure, and `differences.md` Update
|
||||||
|
|
||||||
|
**Files:**
|
||||||
|
- Modify: `docs/plans/2026-02-23-jetstream-remaining-parity-map.md`
|
||||||
|
- Modify: `docs/plans/2026-02-23-jetstream-remaining-parity-verification.md`
|
||||||
|
- Modify: `differences.md`
|
||||||
|
|
||||||
|
**Step 1: Run focused transport+JetStream suites**
|
||||||
|
|
||||||
|
Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStream|FullyQualifiedName~Raft|FullyQualifiedName~Route|FullyQualifiedName~Gateway|FullyQualifiedName~Leaf" -v minimal`
|
||||||
|
Expected: PASS.
|
||||||
|
|
||||||
|
**Step 2: Run full suite**
|
||||||
|
|
||||||
|
Run: `dotnet test -v minimal`
|
||||||
|
Expected: PASS.
|
||||||
|
|
||||||
|
**Step 3: Enforce no JetStream partials in differences**
|
||||||
|
|
||||||
|
Run: `rg -n "## 11\. JetStream|Partial|partial" differences.md`
|
||||||
|
Expected: JetStream section no longer marks remaining entries as partial unless explicitly documented external blockers.
|
||||||
|
|
||||||
|
**Step 4: Update parity evidence rows with exact code+test references**
|
||||||
|
|
||||||
|
```md
|
||||||
|
| $JS.API.STREAM.PEER.REMOVE.* | ClusterControlApiHandlers.HandleStreamPeerRemove | ported | JetStreamClusterControlExtendedApiTests.Peer_remove_and_consumer_stepdown_subjects_return_success_shape |
|
||||||
|
```
|
||||||
|
|
||||||
|
**Step 5: Commit**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git add docs/plans/2026-02-23-jetstream-remaining-parity-map.md docs/plans/2026-02-23-jetstream-remaining-parity-verification.md differences.md
|
||||||
|
git commit -m "docs: close remaining jetstream parity and strict gate evidence"
|
||||||
|
```
|
||||||
|
|
||||||
|
## Dependency Order
|
||||||
|
|
||||||
|
1. Task 1 -> Task 2
|
||||||
|
2. Task 3 -> Task 4 -> Task 5
|
||||||
|
3. Task 6 -> Task 7
|
||||||
|
4. Task 8 -> Task 9
|
||||||
|
5. Task 10 -> Task 11 -> Task 12 -> Task 13 -> Task 14
|
||||||
|
6. Task 15 -> Task 16
|
||||||
|
7. Task 17
|
||||||
|
|
||||||
|
## Executor Notes
|
||||||
|
|
||||||
|
- Use Go references while implementing each task:
|
||||||
|
- `golang/nats-server/server/jetstream_api.go`
|
||||||
|
- `golang/nats-server/server/jetstream.go`
|
||||||
|
- `golang/nats-server/server/stream.go`
|
||||||
|
- `golang/nats-server/server/consumer.go`
|
||||||
|
- `golang/nats-server/server/raft.go`
|
||||||
|
- `golang/nats-server/server/route.go`
|
||||||
|
- `golang/nats-server/server/gateway.go`
|
||||||
|
- `golang/nats-server/server/leafnode.go`
|
||||||
|
- Keep behavior claims test-backed; do not update parity status based only on type signatures or route registration.
|
||||||
Reference in New Issue
Block a user