diff --git a/docs/plans/2026-02-23-jetstream-deep-operational-parity-plan.md b/docs/plans/2026-02-23-jetstream-deep-operational-parity-plan.md new file mode 100644 index 0000000..20686bc --- /dev/null +++ b/docs/plans/2026-02-23-jetstream-deep-operational-parity-plan.md @@ -0,0 +1,641 @@ +# JetStream Deep Operational Parity Implementation Plan + +> **For Codex:** REQUIRED SUB-SKILL: Use `executeplan` to implement this plan task-by-task. + +**Goal:** Close remaining deep JetStream operational parity gaps versus Go by hardening runtime semantics, storage durability, RAFT/cluster behavior, and parity documentation accuracy. + +**Architecture:** Execute in strict dependency layers: first codify JetStream truth-matrix assertions, then close stream and consumer runtime semantics, then harden storage durability and RAFT/cluster governance with behavior-real tests, and finally reconcile parity documentation to verified evidence. Treat stale doc claims (including prior `JETSTREAM (internal)` contradictions) as documentation drift that must be validated and corrected. + +**Tech Stack:** .NET 10, C# 14, xUnit 3, Shouldly, NATS server internals, System.Text.Json, System.IO, integration fixtures. + +--- + +**Execution guardrails** +- Use `@test-driven-development` for every task. +- If behavior diverges from expected protocol/consensus semantics, switch to `@systematic-debugging` before further implementation. +- Keep one commit per task. +- Run `@verification-before-completion` before parity closure claims. + +### Task 1: Add JetStream Truth-Matrix Guardrail Tests and Document Drift Detection + +**Files:** +- Create: `tests/NATS.Server.Tests/Parity/JetStreamParityTruthMatrixTests.cs` +- Modify: `tests/NATS.Server.Tests/DifferencesParityClosureTests.cs` +- Modify: `docs/plans/2026-02-23-jetstream-remaining-parity-map.md` + +**Step 1: Write the failing test** + +```csharp +[Fact] +public void Jetstream_parity_rows_require_behavior_test_and_docs_alignment() +{ + var report = JetStreamParityTruthMatrix.Load("differences.md", "docs/plans/2026-02-23-jetstream-remaining-parity-map.md"); + report.DriftRows.ShouldBeEmpty(); +} +``` + +**Step 2: Run test to verify it fails** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamParityTruthMatrixTests|FullyQualifiedName~DifferencesParityClosureTests" -v minimal` +Expected: FAIL with row-level mismatches (summary/table/test evidence drift). + +**Step 3: Write minimal implementation** + +```csharp +public sealed record DriftRow(string Feature, string DifferencesStatus, string EvidenceStatus, string Reason); +public IReadOnlyList DriftRows => _rows.Where(r => r.HasDrift).ToArray(); +``` + +**Step 4: Run test to verify it passes** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamParityTruthMatrixTests|FullyQualifiedName~DifferencesParityClosureTests" -v minimal` +Expected: PASS after matrix + docs are aligned. + +**Step 5: Commit** + +```bash +git add tests/NATS.Server.Tests/Parity/JetStreamParityTruthMatrixTests.cs tests/NATS.Server.Tests/DifferencesParityClosureTests.cs docs/plans/2026-02-23-jetstream-remaining-parity-map.md +git commit -m "test: add jetstream truth-matrix drift guardrails" +``` + +### Task 2: Verify and Lock Internal JetStream Client Parity (`JETSTREAM (internal)`) + +**Files:** +- Modify: `src/NATS.Server/NatsServer.cs` +- Modify: `src/NATS.Server/JetStream/JetStreamService.cs` +- Modify: `tests/NATS.Server.Tests/JetStreamInternalClientTests.cs` +- Create: `tests/NATS.Server.Tests/JetStreamInternalClientRuntimeTests.cs` +- Modify: `differences.md` + +**Step 1: Write the failing test** + +```csharp +[Fact] +public async Task Internal_jetstream_client_is_created_bound_to_sys_account_and_used_by_jetstream_service_lifecycle() +{ + await using var fx = await JetStreamInternalClientFixture.StartAsync(); + fx.JetStreamInternalClientKind.ShouldBe(ClientKind.JetStream); + fx.JetStreamServiceUsesInternalClient.ShouldBeTrue(); +} +``` + +**Step 2: Run test to verify it fails** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamInternalClientTests|FullyQualifiedName~JetStreamInternalClientRuntimeTests" -v minimal` +Expected: FAIL if lifecycle usage assertions are incomplete. + +**Step 3: Write minimal implementation** + +```csharp +_jetStreamInternalClient = new InternalClient(jsClientId, ClientKind.JetStream, _systemAccount); +_jetStreamService = new JetStreamService(options.JetStream, _jetStreamInternalClient); +``` + +**Step 4: Run test to verify it passes** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamInternalClientTests|FullyQualifiedName~JetStreamInternalClientRuntimeTests" -v minimal` +Expected: PASS. + +**Step 5: Commit** + +```bash +git add src/NATS.Server/NatsServer.cs src/NATS.Server/JetStream/JetStreamService.cs tests/NATS.Server.Tests/JetStreamInternalClientTests.cs tests/NATS.Server.Tests/JetStreamInternalClientRuntimeTests.cs differences.md +git commit -m "feat: lock internal jetstream client runtime parity and docs" +``` + +### Task 3: Implement Stream Retention Semantics Parity (`Limits/Interest/WorkQueue`) + +**Files:** +- Modify: `src/NATS.Server/JetStream/StreamManager.cs` +- Modify: `src/NATS.Server/JetStream/Models/StreamConfig.cs` +- Modify: `src/NATS.Server/JetStream/Validation/JetStreamConfigValidator.cs` +- Test: `tests/NATS.Server.Tests/JetStream/JetStreamRetentionRuntimeParityTests.cs` + +**Step 1: Write the failing test** + +```csharp +[Fact] +public async Task Workqueue_and_interest_retention_apply_correct_eviction_rules_under_ack_and_interest_changes() +{ + await using var fx = await JetStreamRetentionFixture.StartAsync(); + var state = await fx.RunRetentionScenarioAsync(); + state.InvariantViolations.ShouldBeEmpty(); +} +``` + +**Step 2: Run test to verify it fails** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamRetentionRuntimeParityTests" -v minimal` +Expected: FAIL while retention policies are simplified. + +**Step 3: Write minimal implementation** + +```csharp +switch (stream.Config.Retention) +{ + case RetentionPolicy.WorkQueue: ApplyWorkQueueRetention(stream); break; + case RetentionPolicy.Interest: ApplyInterestRetention(stream); break; + default: ApplyLimitsRetention(stream); break; +} +``` + +**Step 4: Run test to verify it passes** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamRetentionRuntimeParityTests" -v minimal` +Expected: PASS. + +**Step 5: Commit** + +```bash +git add src/NATS.Server/JetStream/StreamManager.cs src/NATS.Server/JetStream/Models/StreamConfig.cs src/NATS.Server/JetStream/Validation/JetStreamConfigValidator.cs tests/NATS.Server.Tests/JetStream/JetStreamRetentionRuntimeParityTests.cs +git commit -m "feat: implement stream retention runtime parity" +``` + +### Task 4: Harden Stream Runtime Policies (`MaxAge`, `MaxMsgsPer`, `MaxMsgSize`, Dedupe Window) + +**Files:** +- Modify: `src/NATS.Server/JetStream/StreamManager.cs` +- Modify: `src/NATS.Server/JetStream/Publish/PublishPreconditions.cs` +- Modify: `src/NATS.Server/JetStream/Publish/JetStreamPublisher.cs` +- Test: `tests/NATS.Server.Tests/JetStream/JetStreamStreamRuntimePolicyLongRunTests.cs` +- Test: `tests/NATS.Server.Tests/JetStream/JetStreamDedupeWindowParityTests.cs` + +**Step 1: Write the failing test** + +```csharp +[Fact] +public async Task Dedupe_window_expires_entries_and_allows_republish_after_window_boundary() +{ + await using var fx = await JetStreamDedupeFixture.StartAsync(); + var result = await fx.PublishAcrossWindowBoundaryAsync(); + result.SecondPublishAcceptedAfterWindow.ShouldBeTrue(); +} +``` + +**Step 2: Run test to verify it fails** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamStreamRuntimePolicyLongRunTests|FullyQualifiedName~JetStreamDedupeWindowParityTests" -v minimal` +Expected: FAIL for long-run timing and dedupe edge cases. + +**Step 3: Write minimal implementation** + +```csharp +_preconditions.TrimOlderThan(stream.Config.DuplicateWindowMs); +if (!_preconditions.CheckExpectedLastSeq(opts.ExpectedLastSeq, state.LastSeq)) return Error(10071); +``` + +**Step 4: Run test to verify it passes** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamStreamRuntimePolicyLongRunTests|FullyQualifiedName~JetStreamDedupeWindowParityTests" -v minimal` +Expected: PASS. + +**Step 5: Commit** + +```bash +git add src/NATS.Server/JetStream/StreamManager.cs src/NATS.Server/JetStream/Publish/PublishPreconditions.cs src/NATS.Server/JetStream/Publish/JetStreamPublisher.cs tests/NATS.Server.Tests/JetStream/JetStreamStreamRuntimePolicyLongRunTests.cs tests/NATS.Server.Tests/JetStream/JetStreamDedupeWindowParityTests.cs +git commit -m "feat: harden stream runtime policy and dedupe window parity" +``` + +### Task 5: Complete Consumer Deliver Policy and Cursor Semantics Parity + +**Files:** +- Modify: `src/NATS.Server/JetStream/Consumers/PullConsumerEngine.cs` +- Modify: `src/NATS.Server/JetStream/ConsumerManager.cs` +- Modify: `src/NATS.Server/JetStream/Models/ConsumerConfig.cs` +- Test: `tests/NATS.Server.Tests/JetStream/JetStreamConsumerDeliverPolicyLongRunTests.cs` + +**Step 1: Write the failing test** + +```csharp +[Fact] +public async Task Deliver_policy_last_per_subject_and_start_time_resolve_consistent_cursor_under_interleaved_subjects() +{ + await using var fx = await ConsumerDeliverPolicyFixture.StartAsync(); + var cursor = await fx.ResolveCursorAsync(); + cursor.InvariantViolations.ShouldBeEmpty(); +} +``` + +**Step 2: Run test to verify it fails** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamConsumerDeliverPolicyLongRunTests" -v minimal` +Expected: FAIL on long-run cursor correctness. + +**Step 3: Write minimal implementation** + +```csharp +DeliverPolicy.LastPerSubject => await ResolveLastPerSubjectAsync(stream, config, ct), +DeliverPolicy.ByStartTime => await ResolveByStartTimeAsync(stream, config.OptStartTimeUtc!.Value, ct), +``` + +**Step 4: Run test to verify it passes** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamConsumerDeliverPolicyLongRunTests" -v minimal` +Expected: PASS. + +**Step 5: Commit** + +```bash +git add src/NATS.Server/JetStream/Consumers/PullConsumerEngine.cs src/NATS.Server/JetStream/ConsumerManager.cs src/NATS.Server/JetStream/Models/ConsumerConfig.cs tests/NATS.Server.Tests/JetStream/JetStreamConsumerDeliverPolicyLongRunTests.cs +git commit -m "feat: complete consumer deliver policy cursor parity" +``` + +### Task 6: Complete Ack/Redelivery/Backoff State-Machine Parity + +**Files:** +- Modify: `src/NATS.Server/JetStream/Consumers/AckProcessor.cs` +- Modify: `src/NATS.Server/JetStream/Consumers/PullConsumerEngine.cs` +- Modify: `src/NATS.Server/JetStream/ConsumerManager.cs` +- Test: `tests/NATS.Server.Tests/JetStream/JetStreamAckRedeliveryStateMachineTests.cs` + +**Step 1: Write the failing test** + +```csharp +[Fact] +public async Task Ack_all_and_backoff_redelivery_follow_monotonic_floor_and_max_deliver_rules() +{ + await using var fx = await AckStateMachineFixture.StartAsync(); + var report = await fx.RunAsync(); + report.InvariantViolations.ShouldBeEmpty(); +} +``` + +**Step 2: Run test to verify it fails** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamAckRedeliveryStateMachineTests" -v minimal` +Expected: FAIL with floor/backoff/max-deliver edge mismatches. + +**Step 3: Write minimal implementation** + +```csharp +if (ackPolicy == AckPolicy.All) _state.AdvanceFloor(sequence); +if (deliveryAttempt > config.MaxDeliver) _state.Terminate(sequence); +``` + +**Step 4: Run test to verify it passes** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamAckRedeliveryStateMachineTests" -v minimal` +Expected: PASS. + +**Step 5: Commit** + +```bash +git add src/NATS.Server/JetStream/Consumers/AckProcessor.cs src/NATS.Server/JetStream/Consumers/PullConsumerEngine.cs src/NATS.Server/JetStream/ConsumerManager.cs tests/NATS.Server.Tests/JetStream/JetStreamAckRedeliveryStateMachineTests.cs +git commit -m "feat: complete consumer ack/redelivery state-machine parity" +``` + +### Task 7: Harden Flow Control, Rate Limiting, and Replay Timing Parity + +**Files:** +- Modify: `src/NATS.Server/JetStream/Consumers/PushConsumerEngine.cs` +- Modify: `src/NATS.Server/JetStream/Consumers/PullConsumerEngine.cs` +- Modify: `src/NATS.Server/JetStream/ConsumerManager.cs` +- Test: `tests/NATS.Server.Tests/JetStream/JetStreamFlowControlReplayTimingTests.cs` + +**Step 1: Write the failing test** + +```csharp +[Fact] +public async Task Push_flow_control_and_rate_limit_frames_follow_expected_timing_order_under_burst_load() +{ + await using var fx = await FlowReplayFixture.StartAsync(); + var trace = await fx.CollectFrameTimelineAsync(); + trace.OrderViolations.ShouldBeEmpty(); +} +``` + +**Step 2: Run test to verify it fails** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamFlowControlReplayTimingTests" -v minimal` +Expected: FAIL with timing/order drift. + +**Step 3: Write minimal implementation** + +```csharp +if (config.FlowControl && ShouldEmitFlowControl(nowUtc)) EnqueueFlowControl(); +if (config.ReplayPolicy == ReplayPolicy.Original) await DelayFromOriginalDeltaAsync(prev, current, ct); +``` + +**Step 4: Run test to verify it passes** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamFlowControlReplayTimingTests" -v minimal` +Expected: PASS. + +**Step 5: Commit** + +```bash +git add src/NATS.Server/JetStream/Consumers/PushConsumerEngine.cs src/NATS.Server/JetStream/Consumers/PullConsumerEngine.cs src/NATS.Server/JetStream/ConsumerManager.cs tests/NATS.Server.Tests/JetStream/JetStreamFlowControlReplayTimingTests.cs +git commit -m "feat: harden consumer flow control and replay timing parity" +``` + +### Task 8: Replace FileStore Hook-Level Blocking With Durable Block/Index Semantics + +**Files:** +- Modify: `src/NATS.Server/JetStream/Storage/FileStoreBlock.cs` +- Modify: `src/NATS.Server/JetStream/Storage/FileStore.cs` +- Modify: `src/NATS.Server/JetStream/Storage/FileStoreOptions.cs` +- Modify: `src/NATS.Server/JetStream/Storage/IStreamStore.cs` +- Test: `tests/NATS.Server.Tests/JetStream/JetStreamFileStoreDurabilityParityTests.cs` + +**Step 1: Write the failing test** + +```csharp +[Fact] +public async Task File_store_recovers_block_index_map_after_restart_without_full_log_scan() +{ + await using var fx = await FileStoreDurabilityFixture.StartAsync(); + var result = await fx.ReopenAndVerifyIndexRecoveryAsync(); + result.FullScanRequired.ShouldBeFalse(); +} +``` + +**Step 2: Run test to verify it fails** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamFileStoreDurabilityParityTests" -v minimal` +Expected: FAIL due current rewrite/full-scan style behavior. + +**Step 3: Write minimal implementation** + +```csharp +PersistBlockIndexManifest(_manifestPath, _blockIndex); +LoadBlockIndexManifestOnStartup(); +``` + +**Step 4: Run test to verify it passes** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamFileStoreDurabilityParityTests" -v minimal` +Expected: PASS. + +**Step 5: Commit** + +```bash +git add src/NATS.Server/JetStream/Storage/FileStoreBlock.cs src/NATS.Server/JetStream/Storage/FileStore.cs src/NATS.Server/JetStream/Storage/FileStoreOptions.cs src/NATS.Server/JetStream/Storage/IStreamStore.cs tests/NATS.Server.Tests/JetStream/JetStreamFileStoreDurabilityParityTests.cs +git commit -m "feat: implement durable filestore block and index parity" +``` + +### Task 9: Harden FileStore Compression/Encryption Semantics to Production-Like Contracts + +**Files:** +- Modify: `src/NATS.Server/JetStream/Storage/FileStore.cs` +- Modify: `src/NATS.Server/JetStream/Storage/FileStoreOptions.cs` +- Test: `tests/NATS.Server.Tests/JetStream/JetStreamFileStoreCompressionEncryptionParityTests.cs` + +**Step 1: Write the failing test** + +```csharp +[Fact] +public async Task Compression_and_encryption_roundtrip_is_versioned_and_detects_wrong_key_corruption() +{ + await using var fx = await FileStoreCryptoFixture.StartAsync(); + var report = await fx.VerifyCryptoContractsAsync(); + report.ContractViolations.ShouldBeEmpty(); +} +``` + +**Step 2: Run test to verify it fails** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamFileStoreCompressionEncryptionParityTests" -v minimal` +Expected: FAIL because current XOR/deflate stubs are insufficient. + +**Step 3: Write minimal implementation** + +```csharp +var sealedPayload = _aead.Seal(nonce, plaintext, associatedData); +var compressed = S2Codec.Compress(plaintext); +``` + +**Step 4: Run test to verify it passes** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamFileStoreCompressionEncryptionParityTests" -v minimal` +Expected: PASS. + +**Step 5: Commit** + +```bash +git add src/NATS.Server/JetStream/Storage/FileStore.cs src/NATS.Server/JetStream/Storage/FileStoreOptions.cs tests/NATS.Server.Tests/JetStream/JetStreamFileStoreCompressionEncryptionParityTests.cs +git commit -m "feat: harden filestore compression and encryption parity" +``` + +### Task 10: Implement RAFT Append/Commit Semantics Beyond Hook-Level Replication + +**Files:** +- Modify: `src/NATS.Server/Raft/RaftNode.cs` +- Modify: `src/NATS.Server/Raft/RaftReplicator.cs` +- Modify: `src/NATS.Server/Raft/RaftLog.cs` +- Modify: `src/NATS.Server/Raft/RaftRpcContracts.cs` +- Test: `tests/NATS.Server.Tests/Raft/RaftAppendCommitParityTests.cs` + +**Step 1: Write the failing test** + +```csharp +[Fact] +public async Task Leader_commits_only_after_quorum_and_rejects_conflicting_log_index_term_sequences() +{ + await using var cluster = await RaftAppendFixture.StartAsync(); + var report = await cluster.RunCommitConflictScenarioAsync(); + report.InvariantViolations.ShouldBeEmpty(); +} +``` + +**Step 2: Run test to verify it fails** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~RaftAppendCommitParityTests" -v minimal` +Expected: FAIL with conflict/quorum gaps. + +**Step 3: Write minimal implementation** + +```csharp +if (!Log.MatchesPrev(prevLogIndex, prevLogTerm)) return AppendRejected(); +if (acks + 1 >= quorum) CommitTo(index); +``` + +**Step 4: Run test to verify it passes** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~RaftAppendCommitParityTests" -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/RaftRpcContracts.cs tests/NATS.Server.Tests/Raft/RaftAppendCommitParityTests.cs +git commit -m "feat: implement raft append/commit operational parity" +``` + +### Task 11: Implement RAFT Heartbeat, NextIndex Backtracking, Snapshot Catch-up, Membership Changes + +**Files:** +- Modify: `src/NATS.Server/Raft/RaftTransport.cs` +- Modify: `src/NATS.Server/Raft/RaftNode.cs` +- Modify: `src/NATS.Server/Raft/RaftReplicator.cs` +- Modify: `src/NATS.Server/Raft/RaftSnapshotStore.cs` +- Test: `tests/NATS.Server.Tests/Raft/RaftOperationalConvergenceParityTests.cs` + +**Step 1: Write the failing test** + +```csharp +[Fact] +public async Task Lagging_follower_converges_via_next_index_backtrack_then_snapshot_install_under_membership_change() +{ + await using var cluster = await RaftConvergenceFixture.StartAsync(); + var result = await cluster.RunLaggingFollowerScenarioAsync(); + result.Converged.ShouldBeTrue(); +} +``` + +**Step 2: Run test to verify it fails** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~RaftOperationalConvergenceParityTests" -v minimal` +Expected: FAIL for convergence/membership edge behavior. + +**Step 3: Write minimal implementation** + +```csharp +while (!followerAccepted) nextIndex[followerId] = Math.Max(1, nextIndex[followerId] - 1); +if (nextIndex[followerId] <= snapshot.LastIncludedIndex) await transport.InstallSnapshotAsync(...); +``` + +**Step 4: Run test to verify it passes** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~RaftOperationalConvergenceParityTests" -v minimal` +Expected: PASS. + +**Step 5: Commit** + +```bash +git add src/NATS.Server/Raft/RaftTransport.cs src/NATS.Server/Raft/RaftNode.cs src/NATS.Server/Raft/RaftReplicator.cs src/NATS.Server/Raft/RaftSnapshotStore.cs tests/NATS.Server.Tests/Raft/RaftOperationalConvergenceParityTests.cs +git commit -m "feat: implement raft convergence and membership parity" +``` + +### Task 12: Replace JetStream Cluster Governance Placeholders With Consensus-Backed Behavior + +**Files:** +- Modify: `src/NATS.Server/JetStream/Cluster/JetStreamMetaGroup.cs` +- Modify: `src/NATS.Server/JetStream/Cluster/StreamReplicaGroup.cs` +- Modify: `src/NATS.Server/JetStream/Cluster/AssetPlacementPlanner.cs` +- Modify: `src/NATS.Server/NatsServer.cs` +- Test: `tests/NATS.Server.Tests/JetStream/JetStreamClusterGovernanceBehaviorParityTests.cs` + +**Step 1: Write the failing test** + +```csharp +[Fact] +public async Task Meta_group_and_replica_group_apply_consensus_committed_placement_before_stream_transition() +{ + await using var fx = await JetStreamGovernanceFixture.StartAsync(); + var report = await fx.RunPlacementTransitionScenarioAsync(); + report.InvariantViolations.ShouldBeEmpty(); +} +``` + +**Step 2: Run test to verify it fails** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamClusterGovernanceBehaviorParityTests" -v minimal` +Expected: FAIL with placeholder-only governance behavior. + +**Step 3: Write minimal implementation** + +```csharp +var committedPlan = await _metaGroup.CommitPlacementAsync(config, ct); +await _replicaGroup.ApplyCommittedPlacementAsync(committedPlan, ct); +``` + +**Step 4: Run test to verify it passes** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamClusterGovernanceBehaviorParityTests" -v minimal` +Expected: PASS. + +**Step 5: Commit** + +```bash +git add src/NATS.Server/JetStream/Cluster/JetStreamMetaGroup.cs src/NATS.Server/JetStream/Cluster/StreamReplicaGroup.cs src/NATS.Server/JetStream/Cluster/AssetPlacementPlanner.cs src/NATS.Server/NatsServer.cs tests/NATS.Server.Tests/JetStream/JetStreamClusterGovernanceBehaviorParityTests.cs +git commit -m "feat: replace jetstream governance placeholders with committed behavior" +``` + +### Task 13: Harden Cross-Cluster JetStream Runtime Semantics (Not Counter-Level) + +**Files:** +- Modify: `src/NATS.Server/Gateways/GatewayManager.cs` +- Modify: `src/NATS.Server/NatsServer.cs` +- Modify: `src/NATS.Server/JetStream/StreamManager.cs` +- Test: `tests/NATS.Server.Tests/JetStream/JetStreamCrossClusterBehaviorParityTests.cs` + +**Step 1: Write the failing test** + +```csharp +[Fact] +public async Task Cross_cluster_jetstream_replication_propagates_committed_stream_state_not_just_forward_counter() +{ + await using var fx = await JetStreamCrossClusterFixture.StartAsync(); + var report = await fx.RunReplicationScenarioAsync(); + report.StateDivergence.ShouldBeFalse(); +} +``` + +**Step 2: Run test to verify it fails** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamCrossClusterBehaviorParityTests" -v minimal` +Expected: FAIL while cross-cluster behavior remains shallow. + +**Step 3: Write minimal implementation** + +```csharp +await _gatewayManager.ForwardJetStreamClusterMessageAsync(committedEvent, ct); +ApplyRemoteCommittedStreamEvent(committedEvent); +``` + +**Step 4: Run test to verify it passes** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamCrossClusterBehaviorParityTests" -v minimal` +Expected: PASS. + +**Step 5: Commit** + +```bash +git add src/NATS.Server/Gateways/GatewayManager.cs src/NATS.Server/NatsServer.cs src/NATS.Server/JetStream/StreamManager.cs tests/NATS.Server.Tests/JetStream/JetStreamCrossClusterBehaviorParityTests.cs +git commit -m "feat: harden cross-cluster jetstream runtime parity" +``` + +### Task 14: Final Parity Documentation Reconciliation and Verification Evidence + +**Files:** +- Modify: `differences.md` +- Modify: `docs/plans/2026-02-23-jetstream-remaining-parity-map.md` +- Modify: `docs/plans/2026-02-23-jetstream-remaining-parity-verification.md` + +**Step 1: Write the failing test** + +```csharp +[Fact] +public void Jetstream_differences_notes_have_no_contradictions_against_status_table_and_truth_matrix() +{ + var report = JetStreamParityTruthMatrix.Load("differences.md", "docs/plans/2026-02-23-jetstream-remaining-parity-map.md"); + report.Contradictions.ShouldBeEmpty(); +} +``` + +**Step 2: Run test to verify it fails** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStreamParityTruthMatrixTests|FullyQualifiedName~DifferencesParityClosureTests" -v minimal` +Expected: FAIL until stale contradictory notes are corrected. + +**Step 3: Write minimal implementation** + +```markdown +### Remaining Explicit Deltas +- None after this deep operational parity cycle; previous contradictory notes removed. +``` + +**Step 4: Run verification gates** + +Run: `dotnet test tests/NATS.Server.Tests --filter "FullyQualifiedName~JetStream|FullyQualifiedName~Raft|FullyQualifiedName~Gateway|FullyQualifiedName~Leaf|FullyQualifiedName~Route|FullyQualifiedName~DifferencesParityClosureTests|FullyQualifiedName~JetStreamParityTruthMatrixTests" -v minimal` +Expected: PASS. + +Run: `dotnet test -v minimal` +Expected: PASS. + +**Step 5: Commit** + +```bash +git add differences.md docs/plans/2026-02-23-jetstream-remaining-parity-map.md docs/plans/2026-02-23-jetstream-remaining-parity-verification.md +git commit -m "docs: reconcile jetstream deep parity evidence and status" +```