Sourcing streams between hub and leaf instances (both directions)

[cbrake]

I have been experimenting with Jetstream for use as an IoT data store. The idea is that leaf or hub nodes may come and go, but when things connect again, all the streams sync up.

This example attempts to synchronize streams in two directions.

https://github.com/simpleiot/nats-exp/blob/main/bi-directional-sync/main.go

[cbrake]

added code that shuts down the leaf node, and the leaf stream is still available on the hub instance. Then I power the leaf back up, and publish 5 more messages. On the leaf node, I now see 10 messages (5 before and 5 after cycle leaf node). But on the hub node, I see 15 messages I the sourced stream from the leaf node -- the 2nd set of 5 gets published twice. How is this happening?

To reproduce test:

git clone https://github.com/simpleiot/nats-exp.git
cd nats-exp
go run bi-directional-sync/main.go

[cbrake@quark nats-exp]$ go run bi-directional-sync/main.go 
2023/12/14 18:07:05 Number of stream messages:  4
2023/12/14 18:07:05 Number of hub stream messages from leaf node:  4
2023/12/14 18:07:06 Number of hub->leaf sourced stream messages:  4
2023/12/14 18:07:06 Number of leaf stream messages:  5
2023/12/14 18:07:06 Number of leaf->hub sourced stream messages:  5
2023/12/14 18:07:06 Number of stream messages:hub:NODES-LEAF:5
2023/12/14 18:07:06 Number of hub stream messages:  8
Error leaf counting hub sourced stream messages after leaf server powered back up:  Returned wrong number of messages, expected: 8, got: 4
2023/12/14 18:07:11 Number of leaf stream messages:  10
2023/12/14 18:07:11 Number of stream messages:leaf:NODES-LEAF:10
Message 1: 22
Message 2: 23
Message 3: 24
Message 4: 25
Message 5: 26
Message 6: 27
Message 7: 28
Message 8: 29
Message 9: 30
Message 10: 31
[START DUPLICATED MSGS]
Message 11: 27
Message 12: 28
Message 13: 29
Message 14: 30
Message 15: 31
[END DUPLICATED MESSAGES]
Error hub counting leaf stream messages after leaf server shut down:  Returned wrong number of messages, expected: 10, got: 15
2023/12/14 18:07:16 Number of stream messages:leaf:NODES-LEAF:10
Message 1: 22
Message 2: 23
Message 3: 24
Message 4: 25
Message 5: 26
Message 6: 27
Message 7: 28
Message 8: 29
Message 9: 30
Message 10: 31

[cbrake]

Just added code to push messages to hub while leaf is down, and when leaf reconnects, I don't see the messages on the sourced stream on the leaf node -- probably doing something wrong, but out of time today.

[jnmoyne]

It looks like both the stream on the hub and the one on the LN, are listening to the same subjects. This means that even though the two streams are in different JS domains, as they are listening on the same Core NATS subject if the Core NATS messages on those subject(s) are allowed to flow between the hub and the LNs then both streams will store the message: that could explain duplicated messages.

[cbrake]

Thanks! -- that makes sense now. I did not realize a stream could listen to local subjects as well as source other streams -- that is pretty neat!

The following change fixed the issue:

@@ -333,12 +330,12 @@ func sourceStream(srv *server.Server, domain, stream, sourceDomain, subject stri
        }
 
        cfg := jetstream.StreamConfig{
-               Name:     stream,
-               Subjects: []string{subject},
+               Name: stream,
                Sources: []*jetstream.StreamSource{
                        {
-                               Name:   stream,
-                               Domain: sourceDomain,
+                               Name:          stream,
+                               Domain:        sourceDomain,
+                               FilterSubject: subject,
                        },
                },
``

[cbrake]

I cleaned up this code and increased the timeout, now when I shutdown the leaf node and start it back up, the messages that were published on the hub while it was down sync after about 6s (I had the timeout set to 5 previously, so that appears to be why it did not work).

This is very neat -- everything is now working as expected. One more step forward to a Jetstream store for SIOT.

Log of current program:

[cbrake@quark nats-exp]$ go run ./bi-directional-sync/
=====================================================
       Create stream on hub and source to leaf
=====================================================
* Create stream: server:hub domain:hub stream:NODES-HUB subject:n.hub.>
* Publish: hub -> n.hub.123.value -> 12 13 14 15 
* Stream count: server:hub domain:hub stream:NODES-HUB count:4
* Stream count: server:leaf domain:hub stream:NODES-HUB count:4
* Source stream: server:leaf domain:leaf stream:NODES-HUB source-domain:hub subject:n.hub.>
* Stream count: server:leaf domain:leaf stream:NODES-HUB count:4
=====================================================
       Create stream on leaf and source to hub
=====================================================
* Create stream: server:leaf domain:leaf stream:NODES-LEAF subject:n.leaf.>
* Publish: leaf -> n.leaf.456.value -> 22 23 24 25 26 
* Source stream: server:hub domain:hub stream:NODES-LEAF source-domain:leaf subject:n.leaf.>
* Stream count: server:hub domain:hub stream:NODES-LEAF count:5
=====================================================
       Shut down leaf node
=====================================================
* Stream count: server:hub domain:hub stream:NODES-LEAF count:5
* Publish: hub -> n.hub.123.value -> 16 17 18 19 
* Stream count: server:hub domain:hub stream:NODES-HUB count:8
=====================================================
       Start up leaf node
=====================================================
* Stream count: server:leaf domain:leaf stream:NODES-HUB count:8
* Publish: hub -> n.hub.123.value -> 20 21 22 23 
* Stream count: server:leaf domain:leaf stream:NODES-HUB count:12
=====================================================
       Publish more messages to leaf
=====================================================
* Publish: leaf -> n.leaf.456.value -> 27 28 29 30 31 
* Stream count: server:leaf domain:leaf stream:NODES-LEAF count:10
* Stream count: server:leaf domain:leaf stream:NODES-LEAF count:10
* Stream count: server:hub domain:hub stream:NODES-LEAF count:10