Aggregate KV?

[uli42]

Hello,

I have a non-functioning setup and I am not really understanding what's wrong.

I have watched this video about aggregating streams from multiple site into one stream: https://github.com/nats-io/jetstream-leaf-nodes-demo#connect-streams-cross-accounts respectively https://www.youtube.com/watch?v=0MkS_S7lyHk which both describe how to aggregate streams.

Now our scenario is this: We have multiple (remote) sites producing data and want to aggregate those into a central nats instance. But we want to use KV for this. For the sake of simplicity I am using two sites (clusters) only for now, which have the same accounts/users. There a leafnode connection between the two. I have done the following:

On site1:

$ nats kv add --history 5 --replicas 3 kv1  --storage file --ttl 14d
$ nats kv add --history 5 --replicas 3 kv2  --storage file --ttl 14d

On the central site:

nats kv add --history 5 --replicas 3 kv1_mirror  --storage file --ttl 14d --mirror kv1  --mirror-domain site1
nats kv add --history 5 --replicas 3 kv2_mirror  --storage file --ttl 14d --mirror kv2  --mirror-domain site1

$  nats kv info kv1_mirror
Information for Key-Value Store Bucket kv1_mirror created 2026-02-02 19:28:05

Configuration:

            Bucket Name: kv1_mirror
           History Kept: 5
          Values Stored: 0
             Compressed: false
  Per-Key TTL Supported: false
     Backing Store Kind: JetStream
            Bucket Size: 54 B
    Maximum Bucket Size: unlimited
     Maximum Value Size: unlimited
            Maximum Age: 14d0h0m0s
       JetStream Stream: KV_kv1_mirror
                Storage: File

Mirror Information:

          Origin Bucket: kv1
           External API: $JS.site1.API
              Last Seen: 35ms
                    Lag: 0

Cluster Information:

                   Name: nats-test
                 Leader: nats-test-1
                Replica: nats-test-0, current, seen 951ms ago
                Replica: nats-test-2, current, seen 950ms ago

(kv2_mirror similar)

Whenever I put a key into kv1 or kv2 on site1 I can see that key in kv1_mirror resp. kv2_mirror on the central site. Fine!

Now I create an aggregate bucket on the central site:

$ nats kv add --history 5 --replicas 3 kv_aggregate  --storage file --ttl 14d --source kv1_mirror --source kv2_mirror
$ nats kv info kv_aggregate
Information for Key-Value Store Bucket kv_aggregate created 2026-02-02 19:28:43

Configuration:

            Bucket Name: kv_aggregate
           History Kept: 5
          Values Stored: 0
             Compressed: false
  Per-Key TTL Supported: false
     Backing Store Kind: JetStream
            Bucket Size: 140 B
    Maximum Bucket Size: unlimited
     Maximum Value Size: unlimited
            Maximum Age: 14d0h0m0s
       JetStream Stream: KV_kv_aggregate
                Storage: File

Sources Information:

          Source Bucket: kv1_mirror
              Last Seen: 167ms
                    Lag: 0
          Source Bucket: kv2_mirror
              Last Seen: 76ms
                    Lag: 0

Cluster Information:

                   Name: nats-test
                 Leader: nats-test-1
                Replica: nats-test-0, current, seen 67ms ago
                Replica: nats-test-2, current, seen 67ms ago

But kv_aggregate is always empty. I would have expected to see all the keys of either kv1_mirror or kv2_mirror, but nothing.

I am pretty sure I have some misunderstanding somewhere. Can someone please explain to me why kv_aggregate does not see any messages? How should this setup look to work as intended (all keys in one bucket on the central site)?

[jnmoyne]

You need to do a little bit of subject transformation as you source into your aggregated stream, as the name of the bucket is part of the subject of the messages when stored in the stream underlying a KV bucket.

Example of the StreamConfig for a bucket C that sources from KV buckets A and B:

{
  "name": "KV_C",
  "subjects": [
    "$KV.C.>"
  ],
  "retention": "limits",
  "max_consumers": -1,
  "max_msgs_per_subject": 1,
  "max_msgs": -1,
  "max_bytes": -1,
  "max_age": 0,
  "max_msg_size": -1,
  "storage": "file",
  "discard": "new",
  "num_replicas": 1,
  "duplicate_window": 120000000000,
  "placement": {
    "cluster": ""
  },
  "sources": [
    {
      "name": "KV_A",
      "filter_subject": "$KV.A.>",
      "subject_transform_dest": "$KV.C.>"
    },
    {
      "name": "KV_B",
      "filter_subject": "$KV.B.>",
      "subject_transform_dest": "$KV.C.>"
    }
  ],
  "sealed": false,
  "deny_delete": true,
  "deny_purge": false,
  "allow_rollup_hdrs": true,
  "allow_direct": true,
  "mirror_direct": false
}

(see #1200 for reference)
Also nats does that automatically for you (e.g. when you do nats kv add xxx --source yyy), and some client libraries also do (e.g. see

nats.go/jetstream/kv.go

Line 246 in e870594

Mirror *StreamSource `json:"mirror,omitempty"`

)

[uli42]

Thanks for your answer. However, you state "nats does this automatically for you".

So checking the underlying stream of the aggregate KV bucket I indeed see:

Source Information:

                   Stream Name: KV_kv1_mirror
  Subject Filter and Transform: $KV.kv1_mirror.> to $KV.kv_aggregate.>
                           Lag: 0
                     Last Seen: 255ms

                   Stream Name: KV_kv2_mirror
  Subject Filter and Transform: $KV.kv2_mirror.> to $KV.kv_aggregate.>
                           Lag: 0
                     Last Seen: 941ms

But still: no messages in kv_aggregate

So I have changed the settings like this

@@ -20,7 +20,7 @@
         "name": "KV_kv1_mirror",
         "subject_transforms": [
           {
-            "src": "$KV.kv1_mirror.\u003e",
+            "src": "$KV.kv1.\u003e",
             "dest": "$KV.kv_aggregate.\u003e"
           }
         ]
@@ -29,7 +29,7 @@
         "name": "KV_kv2_mirror",
         "subject_transforms": [
           {
-            "src": "$KV.kv2_mirror.\u003e",
+            "src": "$KV.kv2.\u003e",
             "dest": "$KV.kv_aggregate.\u003e"
           }
         ]

and it now works! Thanks again!

( I think it should also be possible to do the transformation directly on the mirrors but this had not worked at first attempt, so I dropped the idea for now)

[uli42]

One more question: when I now delete a KV in the aggregate it vanishes there but not in the two mirrors where the aggregate is sourcing from. Can this be achieved somehow?

[jnmoyne]

No, mirroring and sourcing only propagate new messages, not deletes.