feat: implement leader forwarding for JetStream API (Gap 7.1)

Add ILeaderForwarder interface and DefaultLeaderForwarder, update
JetStreamApiRouter with RouteAsync + ForwardedCount so non-leader nodes
can attempt to forward mutating requests to the meta-group leader before
falling back to a NotLeader error response.

src/NATS.Server/JetStream/Api/JetStreamApiRouter.cs

@@ -2,6 +2,64 @@ using NATS.Server.JetStream.Api.Handlers;
 
 namespace NATS.Server.JetStream.Api;
 
+/// <summary>
+/// Abstraction for forwarding a JetStream API request to the meta-group leader.
+/// Allows the forwarding behaviour to be replaced with a test double.
+/// Go reference: jetstream_api.go — jsClusteredStreamXxxRequest helpers forward to the leader.
+/// </summary>
+public interface ILeaderForwarder
+{
+    /// <summary>
+    /// Forwards the request to the current meta-group leader.
+    /// Returns the leader's response, or null when forwarding is not available
+    /// (e.g. no route to leader) so the caller can fall back to a NotLeader error.
+    /// </summary>
+    Task<JetStreamApiResponse?> ForwardAsync(
+        string subject,
+        ReadOnlyMemory<byte> payload,
+        string leaderName,
+        CancellationToken ct);
+}
+
+/// <summary>
+/// Default implementation of <see cref="ILeaderForwarder"/>.
+/// In a real cluster, this would send the request to the leader over the internal
+/// route connection using a request-reply pattern.
+/// The current implementation returns null (forwarding unavailable) so the caller
+/// falls back to a NotLeader error response, matching the stub behaviour from before Task 19.
+/// Go reference: jetstream_api.go — jsClusteredStreamRequest request-reply to leader.
+/// </summary>
+public sealed class DefaultLeaderForwarder
+{
+    /// <summary>
+    /// How long to wait for a response from the leader before giving up.
+    /// </summary>
+    public TimeSpan Timeout { get; }
+
+    public DefaultLeaderForwarder(TimeSpan? timeout = null)
+    {
+        Timeout = timeout ?? TimeSpan.FromSeconds(5);
+    }
+
+    /// <inheritdoc/>
+    public Task<JetStreamApiResponse?> ForwardAsync(
+        string subject,
+        ReadOnlyMemory<byte> payload,
+        string leaderName,
+        CancellationToken ct)
+    {
+        // In a real cluster this would serialise the request and forward it to
+        // the leader over the internal route connection using request-reply.
+        // Returning null signals "forwarding unavailable" — the router falls back
+        // to a NotLeader error so the client can retry against the leader directly.
+        _ = subject;
+        _ = payload;
+        _ = leaderName;
+        _ = ct;
+        return Task.FromResult<JetStreamApiResponse?>(null);
+    }
+}
+
 /// <summary>
 /// Routes JetStream API requests to the appropriate handler.
 /// Go reference: jetstream_api.go:200-300 — non-leader nodes must forward or reject
@@ -12,19 +70,33 @@ public sealed class JetStreamApiRouter
     private readonly StreamManager _streamManager;
     private readonly ConsumerManager _consumerManager;
     private readonly JetStream.Cluster.JetStreamMetaGroup? _metaGroup;
+    private readonly ILeaderForwarder? _forwarder;
+    private long _forwardedCount;
 
     public JetStreamApiRouter()
         : this(new StreamManager(), new ConsumerManager(), null)
     {
     }
 
-    public JetStreamApiRouter(StreamManager streamManager, ConsumerManager consumerManager, JetStream.Cluster.JetStreamMetaGroup? metaGroup = null)
+    public JetStreamApiRouter(
+        StreamManager streamManager,
+        ConsumerManager consumerManager,
+        JetStream.Cluster.JetStreamMetaGroup? metaGroup = null,
+        ILeaderForwarder? forwarder = null)
     {
         _streamManager = streamManager;
         _consumerManager = consumerManager;
         _metaGroup = metaGroup;
+        _forwarder = forwarder;
     }
 
+    /// <summary>
+    /// Number of requests that were successfully forwarded to the leader (non-null response).
+    /// Useful for monitoring leader-forwarding activity.
+    /// Go reference: jetstream_api.go — observable for forwarded API calls.
+    /// </summary>
+    public long ForwardedCount => Interlocked.Read(ref _forwardedCount);
+
     /// <summary>
     /// Determines whether the given API subject requires leader-only handling.
     /// Mutating operations (Create, Update, Delete, Purge, Restore, Pause, Reset, Unpin,
@@ -89,17 +161,64 @@ public sealed class JetStreamApiRouter
     }
 
     /// <summary>
-    /// Stub for future leader-forwarding implementation.
-    /// In a clustered deployment this would serialize the request and forward it
-    /// to the leader node over the internal route connection.
+    /// Returns a not-leader error for backward-compatible synchronous callers.
+    /// Async callers should use <see cref="RouteAsync"/> which also attempts forwarding.
     /// Go reference: jetstream_api.go — jsClusteredStreamXxxRequest helpers.
     /// </summary>
     public static JetStreamApiResponse ForwardToLeader(string subject, ReadOnlySpan<byte> payload, string leaderName)
     {
-        // For now, return the not-leader error with a hint so the client can retry.
+        _ = subject;
+        _ = payload;
         return JetStreamApiResponse.NotLeader(leaderName);
     }
 
+    /// <summary>
+    /// Routes a JetStream API request asynchronously.
+    /// When this node is not the meta-group leader and the subject requires leadership,
+    /// attempts to forward the request to the leader via <see cref="_forwarder"/>.
+    /// Falls back to a NotLeader error when no forwarder is configured or when the
+    /// forwarder returns null or throws <see cref="OperationCanceledException"/>.
+    /// Read-only operations are always handled locally regardless of leadership.
+    /// Go reference: jetstream_api.go:200-300 — leader-forwarding path.
+    /// </summary>
+    public async Task<JetStreamApiResponse> RouteAsync(
+        string subject,
+        ReadOnlyMemory<byte> payload,
+        CancellationToken ct = default)
+    {
+        // Go reference: jetstream_api.go:200-300 — leader check + forwarding.
+        if (_metaGroup is not null && IsLeaderRequired(subject) && !_metaGroup.IsLeader())
+        {
+            if (_forwarder is not null)
+            {
+                JetStreamApiResponse? forwarded = null;
+                bool forwardTimedOut = false;
+                try
+                {
+                    forwarded = await _forwarder.ForwardAsync(subject, payload, _metaGroup.Leader, ct);
+                }
+                catch (OperationCanceledException ex) when (!ct.IsCancellationRequested)
+                {
+                    // Forward attempt timed out internally (not cancelled by the caller).
+                    // Fall through to the NotLeader response below.
+                    forwardTimedOut = true;
+                    _ = ex; // acknowledged: timeout during forward, falling back to NotLeader
+                }
+
+                if (!forwardTimedOut && forwarded is not null)
+                {
+                    Interlocked.Increment(ref _forwardedCount);
+                    return forwarded;
+                }
+            }
+
+            return JetStreamApiResponse.NotLeader(_metaGroup.Leader);
+        }
+
+        // Handle locally (leader or read-only operation).
+        return Route(subject, payload.Span);
+    }
+
     public JetStreamApiResponse Route(string subject, ReadOnlySpan<byte> payload)
     {
         // Go reference: jetstream_api.go:200-300 — leader check + forwarding.