Clients subscribing to a replicated or sourced stream

[whysi]

Hi,
recently I've started to use NATS JetStream, testing with various topology configurations (clustering+leafnodes).
This is the desired configuration:

• A HUB cluster (3 nodes) on my "main" region, where I centralize various streams
• A LeafNode cluster (3 nodes) on the edge in the "remote_one" region, where a local client publishes in a "timings" stream
• A TEST account on both clusters, jetstream enebled and with full permissions on "timings"
• A mirror stream in the HUB cluster called "timings_mirror", mirroring the "timings" stream on the remote_one region
• A client (.NET application) on the main region should be able to subscribe to the timings_mirror stream

From an infrastructure point of view I'm able to create the clusters, configure the leafnodes connection, create the stream (both timings and timings_mirror) and publish messages using a user in the TEST account.
If I use the NATS Cli I can subscribe to both ends using 2 contexts with the same user, without any problems. I can subscribe to the whole stream or to any of the subjects that I've created in the leaf node without problem, and the stream mirroring works as expected.
But I've a problem doing the same with nats.net: I'm only able to subscribe to the leaf timings stream, and not the replica one.
This is the error I get when I try to subscribe to the stream in the HUB cluster:

NATS.Client.NATSJetStreamClientException: '[SUB-90007] No matching streams for subject

If I try to get the stream info of timings_mirror using GetStreamInfo() I get 0 subjects for the stream, even if I have a number of subjects on the leaf node/original stream.

I don't know if I'm missing a piece of the configuration and/or in the nats.net usage. If needed I can give you more information about my test infrastructure and about the code that I'm using to connect on my application.
I'm asking here because I really don't know if connecting to a mirror (or even a source stream, I've done the same tests with same result) is something that's supported by nats.net or not.
Any help would be really appreciated, thanks in advance!

Regards
Marco

[scottf]

Can you provide some configuration files and some code. The code should specify how you are trying to subscribe. It might be as simple as you needing to give the mirror stream name to the consumer configuration you subscribe with. Also you may need to adjust your jetstream prefix / domain. The .net client is raw, meaning it has all the functionality to support this, but it doesn't do any logic for you regarding understanding your account and domains.
Also would be useful the getStreamInfo call how you did it. To get subjects for the stream, try

StreamInfo GetStreamInfo(string streamName, StreamInfoOptions options);

maybe with

StreamInfoOptions.Builder().WithAllSubjects();

[whysi]

Hi Scott, thanks for your answer.
Sure, here is my main cluster configuration:

http_port: 8222
listen: 4222

jetstream {
  store_dir: /data/nats/
  domain: hub
}

leafnodes {
    listen: 7222
    no_advertise: true
}

cluster {
  name: main_hub
  listen: 0.0.0.0:4248
  authorization {
    user: route_user
    password: routeuserpassword
  }
  routes = [
    nats-route://route_user:routeuserpassword@main_hub_node_1:4248
    nats-route://route_user:routeuserpassword@main_hub_node_2:4248
    nats-route://route_user:routeuserpassword@main_hub_node_3:4248
  ]
}

Leafnode cluster configuration:

http_port: 8111
listen: 4111

jetstream {
    store_dir: /data/nats
    domain: leaf
}

cluster {
  name: leaf_remote_one
  listen: 0.0.0.0:4249
  authorization {
    user: route_user
    password: routeuserpassword
  }
  routes = [
    nats-route://route_user:routeuserpassword@leaf_remote_one_node_1:4249
    nats-route://route_user:routeuserpassword@leaf_remote_one_node_2:4249
    nats-route://route_user:routeuserpassword@leaf_remote_one_node_3:4249
  ]
}

leafnodes {
	no_advertise: true
    remotes = [
		{
			urls: [
            "nats-leaf://admin:adminpassword@main_hub_node_1:7222",
            "nats-leaf://admin:adminpassword@main_hub_node_2:7222",
            "nats-leaf://admin:adminpassword@main_hub_node_3:7222"
            ]
			account: "SYS"
		},
		{
			urls: [
            "nats-leaf://test:testpassword@main_hub_node_1:7222",
            "nats-leaf://test:testpassword@main_hub_node_2:7222",
            "nats-leaf://test:testpassword@main_hub_node_3:7222"
            ]
            account: "ACC"
		},
	]
}

my accounts.conf:

    SYS: {
        users: [{
        	user: "admin",
            password: "adminpassword"
        }]
    },
    ACC: {
        users: [{
        	user: "test",
        	password: "testpassword" 
            },
            { 
            user: "read_only"
            permissions: {
        		publish: {
                    deny: ">"
        		},
                subscribe: {
                	allow: ">"
                }
      		}
    	}],
        jetstream: enabled
    }
}

no_auth_user: "read_only"
system_account: SYS

I've created 2 subject on the "timings" stream (timings.one and timings.two) with 10 example messages each.
The "timings" stream info:

Information for Stream timings created 2023-02-14 10:05:32

             Subjects: timings.one, timings.two
             Replicas: 3
              Storage: File

Options:

            Retention: Limits
     Acknowledgements: true
       Discard Policy: Old
     Duplicate Window: 2m0s
    Allows Msg Delete: true
         Allows Purge: true
       Allows Rollups: false

Limits:

     Maximum Messages: unlimited
  Maximum Per Subject: unlimited
        Maximum Bytes: unlimited
          Maximum Age: unlimited
 Maximum Message Size: unlimited
    Maximum Consumers: unlimited


Cluster Information:

                 Name: leaf_remote_one
               Leader: leaf_remote_one_node_3
              Replica: leaf_remote_one_node_1, current, seen 0.47s ago
              Replica: leaf_remote_one_node_2, current, seen 0.47s ago

State:

             Messages: 20
                Bytes: 1.5 KiB
             FirstSeq: 1 @ 2023-02-14T09:07:51 UTC
              LastSeq: 20 @ 2023-02-14T09:07:55 UTC
     Active Consumers: 0
   Number of Subjects: 2

the "timings_replica" stream info:

Information for Stream timings_mirror created 2023-02-14 10:06:47

             Replicas: 3
              Storage: File

Options:

            Retention: Limits
     Acknowledgements: true
       Discard Policy: Old
     Duplicate Window: 0s
    Allows Msg Delete: true
         Allows Purge: true
       Allows Rollups: false

Limits:

     Maximum Messages: unlimited
  Maximum Per Subject: unlimited
        Maximum Bytes: unlimited
          Maximum Age: unlimited
 Maximum Message Size: unlimited
    Maximum Consumers: unlimited

Limits:

               Mirror: timings, API Prefix: $JS.leaf.API, Delivery Prefix:


Cluster Information:

                 Name: main_hub
               Leader: main_hub_node_2
              Replica: main_hub_node_1, current, seen 0.19s ago
              Replica: main_hub_node_3, current, seen 0.19s ago

Mirror Information:

          Stream Name: timings
                  Lag: 0
            Last Seen: 0.36s
      Ext. API Prefix: $JS.leaf.API

State:

             Messages: 20
                Bytes: 1.5 KiB
             FirstSeq: 1 @ 2023-02-14T09:07:51 UTC
              LastSeq: 20 @ 2023-02-14T09:07:55 UTC
     Active Consumers: 0
   Number of Subjects: 2

This is the example code that I'm using to test the sub to both streams:

for the "timings" stream on leaf_remote_one cluster:

using System;
using System.Diagnostics;
using System.Text;
using NATS.Client;
using NATS.Client.JetStream;


var options = ConnectionFactory.GetDefaultOptions();
options.Name = Guid.NewGuid().ToString();
options.Url = "nats://leaf_remote_one_node_1:4111";
options.User = "test";
options.Password = "testpassword";
options.AllowReconnect = true;

var connection = new ConnectionFactory().CreateConnection(options);

var jetStreamContext = connection.CreateJetStreamContext();
var jetStreamManagementContext = connection.CreateJetStreamManagementContext();
var j = jetStreamManagementContext.GetStreams();

var s = jetStreamManagementContext.GetStreamInfo("timings", new StreamInfoOptions.StreamInfoOptionsBuilder().WithAllSubjects().Build());

PushSubscribeOptions so = PushSubscribeOptions.Builder().Build();

await Task.Delay(TimeSpan.FromSeconds(2));

var sub = jetStreamContext.PushSubscribeSync("timings.>", so);

var count = 0;
while (true)
{
    var jj = sub.NextMessage();
    Console.WriteLine($"\nMessage Received:{count++}");
}

for the "timings_mirror" stream on main_hub cluster:

using System;
using System.Diagnostics;
using System.Text;
using NATS.Client;
using NATS.Client.JetStream;


var options = ConnectionFactory.GetDefaultOptions();
options.Name = Guid.NewGuid().ToString();
options.Url = "nats://main_hub_node_1:4222";
options.User = "test";
options.Password = "testpassword";
options.AllowReconnect = true;

var connection = new ConnectionFactory().CreateConnection(options);

var jetStreamContext = connection.CreateJetStreamContext();
var jetStreamManagementContext = connection.CreateJetStreamManagementContext();
var j = jetStreamManagementContext.GetStreams();

var s = jetStreamManagementContext.GetStreamInfo("timings_mirror", new StreamInfoOptions.StreamInfoOptionsBuilder().WithAllSubjects().Build());

PushSubscribeOptions so = PushSubscribeOptions.Builder().Build();

await Task.Delay(TimeSpan.FromSeconds(2));

var sub = jetStreamContext.PushSubscribeSync("timings_mirror.>", so);

var count = 0;
while (true)
{
    var jj = sub.NextMessage();
    Console.WriteLine($"\nMessage Received:{count++}");
}

These are the streaminfo for the mirror stream:

I don't get why it has 0 subjects (Subject Count = 0) even if in the mirrored stream there are 2 subjects.

If I change the filter of the subject to:
var sub = jetStreamContext.PushSubscribeSync(">", so);
I'm able to get all the messages of the mirror stream.

If I use any other combination of subject filter (like timings.> or timings_mirror.>) I get the error:
NATS.Client.NATSJetStreamClientException: '[SUB-90007] No matching streams for subject.'

Let me know if you need more info/tests.
Thanks
Marco

[scottf]

@whysi This is a ton to digest, I'll look at it when I can. Pleases try setting the stream name in the push subscribe options. Also, for your stream info, it's not clear which server you are connecting to with the cli, so the full cli command would be useful. I just want to be sure the cli isn't using your default server or something.

[whysi]

@scottf thanks for your reply, actually I've done some more testing and setting the stream name in the push subscribe option did the trick.
To give more context for future reference:
when connecting to the leaf cluster I can use this code:

PushSubscribeOptions so = PushSubscribeOptions.Builder().Build();
var sub = jetStreamContext.PushSubscribeSync("timings.one", so);

when connecting to the main/hub cluster this is the code that worked:

PushSubscribeOptions so = PushSubscribeOptions.Builder().WithStream("timings_replica").Build();
var sub = jetStreamContext.PushSubscribeSync("timings.one", so);

I've also tested with a source stream instead of the mirror one and it works the same way.

BTW in my NATS CLI tests I was using 2 different contexts, one pointing to the main cluster and one pointing to the leaf one, each of them pointg directly to one of the nodes in the respective ecluster. I tested also the connection by shutting down the "other" cluster where I was not connecting to, just to be sure that there were no swarm mesh/clustering shenanigans. I was sure that I was connecting to the right environment during mu sub and pub tests.

Thanks again for your help in sorting this out.
Marco