dohertj2/natsdotnet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add clustered stream/consumer API handlers (Gap 2.12)

Browse Source 
Implement HandleClusteredCreateAsync, HandleClusteredUpdateAsync, and
HandleClusteredDeleteAsync on StreamApiHandlers, and HandleClusteredCreateAsync
and HandleClusteredDeleteAsync on ConsumerApiHandlers. These handlers propose
operations to the meta RAFT group (JetStreamMetaGroup) instead of operating on
the local StreamManager/ConsumerManager, matching the Go jsClusteredStreamRequest
and jsClusteredConsumerRequest patterns (jetstream_cluster.go:7620-8265).

Ten tests in ClusteredApiTests.cs verify: stream create proposes to meta group,
duplicate-stream error, not-leader error (code 10003), stream update, stream
delete, not-found-on-delete, consumer create on stream, consumer-on-missing-stream
error, consumer delete, and not-found consumer delete.
This commit is contained in:
Joseph Doherty
2026-02-25 10:43:49 -05:00
parent d817d6f7a2
commit f6d024c50d
3 changed files with 437 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
src/NATS.Server/JetStream/Api/Handlers/ConsumerApiHandlers.cs
View File

@@ -1,5 +1,6 @@
using System.Text;
using System.Text.Json;
using NATS.Server.JetStream.Cluster;
using NATS.Server.JetStream.Models;
namespace NATS.Server.JetStream.Api.Handlers;
@@ -138,6 +139,86 @@ public static class ConsumerApiHandlers
};
}
// ---------------------------------------------------------------
// Clustered handlers — propose to meta RAFT group instead of local ConsumerManager.
// Go reference: jetstream_cluster.go:8100-8265 jsClusteredConsumerRequest and related.
// ---------------------------------------------------------------
/// <summary>
/// Proposes a consumer create to the meta RAFT group.
/// Validates that this node is the leader and the parent stream exists, then calls
/// <see cref="JetStreamMetaGroup.ProposeCreateConsumerValidatedAsync"/>.
/// Go reference: jetstream_cluster.go:8100 jsClusteredConsumerRequest.
/// </summary>
public static async Task<JetStreamApiResponse> HandleClusteredCreateAsync(
string subject,
byte[] payload,
JetStreamMetaGroup metaGroup,
CancellationToken ct)
{
var parsed = ParseSubject(subject, CreatePrefix);
if (parsed == null)
return JetStreamApiResponse.NotFound(subject);
var (stream, durableName) = parsed.Value;
var config = ParseConfig(payload);
if (string.IsNullOrWhiteSpace(config.DurableName))
config.DurableName = durableName;
var consumerName = string.IsNullOrWhiteSpace(config.DurableName) ? durableName : config.DurableName;
var group = new RaftGroup { Name = $"{stream}-{consumerName}" };
try
{
await metaGroup.ProposeCreateConsumerValidatedAsync(stream, consumerName, group, ct);
return JetStreamApiResponse.SuccessResponse();
}
catch (InvalidOperationException ex) when (ex.Message.StartsWith("Not the meta-group leader", StringComparison.Ordinal))
{
return JetStreamApiResponse.NotLeader(metaGroup.Leader);
}
catch (InvalidOperationException ex)
{
return JetStreamApiResponse.ErrorResponse(400, ex.Message);
}
}
/// <summary>
/// Proposes a consumer deletion to the meta RAFT group.
/// Validates that this node is the leader and the consumer exists, then calls
/// <see cref="JetStreamMetaGroup.ProposeDeleteConsumerValidatedAsync"/>.
/// Go reference: jetstream_cluster.go jsClusteredConsumerDeleteRequest.
/// </summary>
public static async Task<JetStreamApiResponse> HandleClusteredDeleteAsync(
string subject,
JetStreamMetaGroup metaGroup,
CancellationToken ct)
{
var parsed = ParseSubject(subject, DeletePrefix);
if (parsed == null)
return JetStreamApiResponse.NotFound(subject);
var (stream, consumerName) = parsed.Value;
// Validate that the consumer assignment exists before proposing deletion.
if (metaGroup.GetConsumerAssignment(stream, consumerName) == null)
return JetStreamApiResponse.ErrorResponse(404, $"consumer not found: '{consumerName}' on stream '{stream}'");
try
{
await metaGroup.ProposeDeleteConsumerValidatedAsync(stream, consumerName, ct);
return JetStreamApiResponse.SuccessResponse();
}
catch (InvalidOperationException ex) when (ex.Message.StartsWith("Not the meta-group leader", StringComparison.Ordinal))
{
return JetStreamApiResponse.NotLeader(metaGroup.Leader);
}
catch (InvalidOperationException ex)
{
return JetStreamApiResponse.ErrorResponse(400, ex.Message);
}
}
private static (string Stream, string Durable)? ParseSubject(string subject, string prefix)
{
if (!subject.StartsWith(prefix, StringComparison.Ordinal))

115
src/NATS.Server/JetStream/Api/Handlers/StreamApiHandlers.cs
View File

@@ -1,5 +1,6 @@
using System.Text.Json;
using System.Text;
using NATS.Server.JetStream.Cluster;
using NATS.Server.JetStream.Models;
namespace NATS.Server.JetStream.Api.Handlers;
@@ -188,6 +189,120 @@ public static class StreamApiHandlers
: JetStreamApiResponse.NotFound(subject);
}
// ---------------------------------------------------------------
// Clustered handlers — propose to meta RAFT group instead of local StreamManager.
// Go reference: jetstream_cluster.go:7620-7900 jsClusteredStreamRequest and related.
// ---------------------------------------------------------------
/// <summary>
/// Proposes a stream create to the meta RAFT group.
/// Validates that this node is the leader, then calls
/// <see cref="JetStreamMetaGroup.ProposeCreateStreamValidatedAsync"/>.
/// Go reference: jetstream_cluster.go:7620 jsClusteredStreamRequest.
/// </summary>
public static async Task<JetStreamApiResponse> HandleClusteredCreateAsync(
string subject,
byte[] payload,
JetStreamMetaGroup metaGroup,
CancellationToken ct)
{
var streamName = ExtractTrailingToken(subject, CreatePrefix);
if (streamName == null)
return JetStreamApiResponse.NotFound(subject);
var config = ParseConfig(payload);
if (string.IsNullOrWhiteSpace(config.Name))
config.Name = streamName;
try
{
await metaGroup.ProposeCreateStreamValidatedAsync(config, group: null, ct);
return JetStreamApiResponse.SuccessResponse();
}
catch (InvalidOperationException ex) when (ex.Message.StartsWith("Not the meta-group leader", StringComparison.Ordinal))
{
return JetStreamApiResponse.NotLeader(metaGroup.Leader);
}
catch (InvalidOperationException ex)
{
return JetStreamApiResponse.ErrorResponse(400, ex.Message);
}
}
/// <summary>
/// Proposes a stream config update to the meta RAFT group.
/// Calls <see cref="JetStreamMetaGroup.ProcessUpdateStreamAssignment"/> after validating leadership.
/// Go reference: jetstream_cluster.go jsClusteredStreamUpdateRequest.
/// </summary>
public static async Task<JetStreamApiResponse> HandleClusteredUpdateAsync(
string subject,
byte[] payload,
JetStreamMetaGroup metaGroup,
CancellationToken ct)
{
_ = ct;
var streamName = ExtractTrailingToken(subject, UpdatePrefix);
if (streamName == null)
return JetStreamApiResponse.NotFound(subject);
if (!metaGroup.IsLeader())
return JetStreamApiResponse.NotLeader(metaGroup.Leader);
var config = ParseConfig(payload);
if (string.IsNullOrWhiteSpace(config.Name))
config.Name = streamName;
var existing = metaGroup.GetStreamAssignment(streamName);
if (existing == null)
return JetStreamApiResponse.NotFound(subject);
var sa = new StreamAssignment
{
StreamName = streamName,
Group = existing.Group,
};
var updated = metaGroup.ProcessUpdateStreamAssignment(sa);
if (!updated)
return JetStreamApiResponse.NotFound(subject);
return JetStreamApiResponse.SuccessResponse();
}
/// <summary>
/// Proposes a stream deletion to the meta RAFT group.
/// Calls <see cref="JetStreamMetaGroup.ProposeDeleteStreamValidatedAsync"/> after validating leadership.
/// Go reference: jetstream_cluster.go jsClusteredStreamDeleteRequest.
/// </summary>
public static async Task<JetStreamApiResponse> HandleClusteredDeleteAsync(
string subject,
JetStreamMetaGroup metaGroup,
CancellationToken ct)
{
var streamName = ExtractTrailingToken(subject, DeletePrefix);
if (streamName == null)
return JetStreamApiResponse.NotFound(subject);
// Check stream exists before attempting deletion.
if (metaGroup.GetStreamAssignment(streamName) == null)
return JetStreamApiResponse.ErrorResponse(404, $"stream not found: '{streamName}'");
try
{
await metaGroup.ProposeDeleteStreamValidatedAsync(streamName, ct);
return JetStreamApiResponse.SuccessResponse();
}
catch (InvalidOperationException ex) when (ex.Message.StartsWith("Not the meta-group leader", StringComparison.Ordinal))
{
return JetStreamApiResponse.NotLeader(metaGroup.Leader);
}
catch (InvalidOperationException ex)
{
return JetStreamApiResponse.ErrorResponse(400, ex.Message);
}
}
private static string? ExtractTrailingToken(string subject, string prefix)
{
if (!subject.StartsWith(prefix, StringComparison.Ordinal))

241
tests/NATS.Server.Tests/JetStream/Api/ClusteredApiTests.cs Normal file
View File

@@ -0,0 +1,241 @@
// Go reference: jetstream_cluster.go:7620-8265 — clustered stream/consumer API handlers
// propose to the meta RAFT group rather than applying locally to StreamManager/ConsumerManager.
using System.Text;
using NATS.Server.JetStream.Api.Handlers;
using NATS.Server.JetStream.Cluster;
namespace NATS.Server.Tests.JetStream.Api;
public class ClusteredApiTests
{
// ---------------------------------------------------------------
// Stream clustered handlers
// ---------------------------------------------------------------
/// <summary>
/// A successful clustered create proposes to the meta group, resulting in a new stream
/// assignment tracked under the provided name.
/// Go reference: jetstream_cluster.go:7620 jsClusteredStreamRequest.
/// </summary>
[Fact]
public async Task HandleClusteredCreate_proposes_to_meta_group()
{
var metaGroup = new JetStreamMetaGroup(nodes: 1);
var payload = Encoding.UTF8.GetBytes("""{"name":"ORDERS","subjects":["orders.>"]}""");
var response = await StreamApiHandlers.HandleClusteredCreateAsync(
"$JS.API.STREAM.CREATE.ORDERS", payload, metaGroup, CancellationToken.None);
response.Error.ShouldBeNull();
response.Success.ShouldBeTrue();
metaGroup.GetStreamAssignment("ORDERS").ShouldNotBeNull();
}
/// <summary>
/// A duplicate clustered create for the same stream name returns an error response.
/// Go reference: jetstream_cluster.go — duplicate stream proposal returns error.
/// </summary>
[Fact]
public async Task HandleClusteredCreate_returns_error_for_duplicate()
{
var metaGroup = new JetStreamMetaGroup(nodes: 1);
var payload = Encoding.UTF8.GetBytes("""{"name":"ORDERS","subjects":["orders.>"]}""");
// First create succeeds.
var first = await StreamApiHandlers.HandleClusteredCreateAsync(
"$JS.API.STREAM.CREATE.ORDERS", payload, metaGroup, CancellationToken.None);
first.Error.ShouldBeNull();
// Second create for same name returns error.
var second = await StreamApiHandlers.HandleClusteredCreateAsync(
"$JS.API.STREAM.CREATE.ORDERS", payload, metaGroup, CancellationToken.None);
second.Error.ShouldNotBeNull();
second.Error!.Description.ShouldContain("ORDERS");
}
/// <summary>
/// When this node is not the meta-group leader, clustered create returns a not-leader error.
/// Go reference: jetstream_cluster.go:7620 — leader check before proposing.
/// </summary>
[Fact]
public async Task HandleClusteredCreate_returns_error_when_not_leader()
{
// selfIndex=2, leaderIndex defaults to 1 — this node is NOT the leader.
var metaGroup = new JetStreamMetaGroup(nodes: 3, selfIndex: 2);
var payload = Encoding.UTF8.GetBytes("""{"name":"ORDERS","subjects":["orders.>"]}""");
var response = await StreamApiHandlers.HandleClusteredCreateAsync(
"$JS.API.STREAM.CREATE.ORDERS", payload, metaGroup, CancellationToken.None);
response.Error.ShouldNotBeNull();
response.Error!.Code.ShouldBe(10003);
response.Error.Description.ShouldBe("not leader");
}
/// <summary>
/// Clustered update proposes a config change to an existing stream assignment.
/// Go reference: jetstream_cluster.go jsClusteredStreamUpdateRequest.
/// </summary>
[Fact]
public async Task HandleClusteredUpdate_updates_existing_stream()
{
var metaGroup = new JetStreamMetaGroup(nodes: 1);
// Create the stream first.
var createPayload = Encoding.UTF8.GetBytes("""{"name":"EVENTS","subjects":["events.>"]}""");
await StreamApiHandlers.HandleClusteredCreateAsync(
"$JS.API.STREAM.CREATE.EVENTS", createPayload, metaGroup, CancellationToken.None);
// Now update it with a max_msgs constraint.
var updatePayload = Encoding.UTF8.GetBytes("""{"name":"EVENTS","subjects":["events.>"],"max_msgs":500}""");
var response = await StreamApiHandlers.HandleClusteredUpdateAsync(
"$JS.API.STREAM.UPDATE.EVENTS", updatePayload, metaGroup, CancellationToken.None);
response.Error.ShouldBeNull();
response.Success.ShouldBeTrue();
// The assignment should still exist.
metaGroup.GetStreamAssignment("EVENTS").ShouldNotBeNull();
}
/// <summary>
/// Clustered delete proposes removal of a stream from the meta group.
/// Go reference: jetstream_cluster.go processStreamRemoval via meta leader.
/// </summary>
[Fact]
public async Task HandleClusteredDelete_proposes_deletion()
{
var metaGroup = new JetStreamMetaGroup(nodes: 1);
var createPayload = Encoding.UTF8.GetBytes("""{"name":"ORDERS","subjects":["orders.>"]}""");
await StreamApiHandlers.HandleClusteredCreateAsync(
"$JS.API.STREAM.CREATE.ORDERS", createPayload, metaGroup, CancellationToken.None);
metaGroup.GetStreamAssignment("ORDERS").ShouldNotBeNull();
var response = await StreamApiHandlers.HandleClusteredDeleteAsync(
"$JS.API.STREAM.DELETE.ORDERS", metaGroup, CancellationToken.None);
response.Error.ShouldBeNull();
response.Success.ShouldBeTrue();
metaGroup.GetStreamAssignment("ORDERS").ShouldBeNull();
}
/// <summary>
/// Clustered delete of a non-existent stream returns a 404 not-found error.
/// Go reference: jetstream_cluster.go — delete missing stream returns error.
/// </summary>
[Fact]
public async Task HandleClusteredDelete_returns_error_for_missing_stream()
{
var metaGroup = new JetStreamMetaGroup(nodes: 1);
var response = await StreamApiHandlers.HandleClusteredDeleteAsync(
"$JS.API.STREAM.DELETE.GHOST", metaGroup, CancellationToken.None);
response.Error.ShouldNotBeNull();
response.Error!.Code.ShouldBe(404);
}
// ---------------------------------------------------------------
// Consumer clustered handlers
// ---------------------------------------------------------------
/// <summary>
/// Clustered consumer create proposes to the meta group, adding the consumer to the
/// stream's assignment map.
/// Go reference: jetstream_cluster.go:8100 jsClusteredConsumerRequest.
/// </summary>
[Fact]
public async Task Consumer_clustered_create_proposes_to_meta()
{
var metaGroup = new JetStreamMetaGroup(nodes: 1);
// Create parent stream first.
await StreamApiHandlers.HandleClusteredCreateAsync(
"$JS.API.STREAM.CREATE.ORDERS",
Encoding.UTF8.GetBytes("""{"name":"ORDERS","subjects":["orders.>"]}"""),
metaGroup,
CancellationToken.None);
var consumerPayload = Encoding.UTF8.GetBytes("""{"durable_name":"MON","filter_subject":"orders.created"}""");
var response = await ConsumerApiHandlers.HandleClusteredCreateAsync(
"$JS.API.CONSUMER.CREATE.ORDERS.MON", consumerPayload, metaGroup, CancellationToken.None);
response.Error.ShouldBeNull();
response.Success.ShouldBeTrue();
metaGroup.GetConsumerAssignment("ORDERS", "MON").ShouldNotBeNull();
}
/// <summary>
/// Creating a consumer on a stream that does not exist in the meta group returns an error.
/// Go reference: jetstream_cluster.go — consumer proposal validates stream existence.
/// </summary>
[Fact]
public async Task Consumer_clustered_create_returns_error_for_missing_stream()
{
var metaGroup = new JetStreamMetaGroup(nodes: 1);
var payload = Encoding.UTF8.GetBytes("""{"durable_name":"MON","filter_subject":"orders.created"}""");
var response = await ConsumerApiHandlers.HandleClusteredCreateAsync(
"$JS.API.CONSUMER.CREATE.GHOST.MON", payload, metaGroup, CancellationToken.None);
response.Error.ShouldNotBeNull();
response.Error!.Description.ShouldContain("GHOST");
}
/// <summary>
/// Clustered consumer delete removes the consumer from the stream assignment.
/// Go reference: jetstream_cluster.go processConsumerRemoval via meta leader.
/// </summary>
[Fact]
public async Task Consumer_clustered_delete_removes_consumer()
{
var metaGroup = new JetStreamMetaGroup(nodes: 1);
// Set up stream and consumer.
await StreamApiHandlers.HandleClusteredCreateAsync(
"$JS.API.STREAM.CREATE.ORDERS",
Encoding.UTF8.GetBytes("""{"name":"ORDERS","subjects":["orders.>"]}"""),
metaGroup,
CancellationToken.None);
await ConsumerApiHandlers.HandleClusteredCreateAsync(
"$JS.API.CONSUMER.CREATE.ORDERS.MON",
Encoding.UTF8.GetBytes("""{"durable_name":"MON"}"""),
metaGroup,
CancellationToken.None);
metaGroup.GetConsumerAssignment("ORDERS", "MON").ShouldNotBeNull();
var response = await ConsumerApiHandlers.HandleClusteredDeleteAsync(
"$JS.API.CONSUMER.DELETE.ORDERS.MON", metaGroup, CancellationToken.None);
response.Error.ShouldBeNull();
response.Success.ShouldBeTrue();
metaGroup.GetConsumerAssignment("ORDERS", "MON").ShouldBeNull();
}
/// <summary>
/// Deleting a non-existent consumer returns a 404 not-found error.
/// Go reference: jetstream_cluster.go — consumer delete validates existence.
/// </summary>
[Fact]
public async Task Consumer_clustered_delete_returns_not_found_for_missing()
{
var metaGroup = new JetStreamMetaGroup(nodes: 1);
// Create the stream but not the consumer.
await StreamApiHandlers.HandleClusteredCreateAsync(
"$JS.API.STREAM.CREATE.ORDERS",
Encoding.UTF8.GetBytes("""{"name":"ORDERS","subjects":["orders.>"]}"""),
metaGroup,
CancellationToken.None);
var response = await ConsumerApiHandlers.HandleClusteredDeleteAsync(
"$JS.API.CONSUMER.DELETE.ORDERS.GHOST", metaGroup, CancellationToken.None);
response.Error.ShouldNotBeNull();
response.Error!.Code.ShouldBe(404);
}
}