Key Value and Object store do not always work with mirrored streams

[NicholasTNG]

What version were you using?

nats-server: v2.10.1
nats-py==2.4.0

What environment was the server running in?

A docker compose setup with separate containers for leaf and remote nats servers (the official images). We have separated jetstream domains with the names "cloud" and "leaf-a". All subjects are synced between the two servers.
The python client is running in a nats-box container.

Is this defect reproducible?

Consider the following code

import nats.aio.client
from nats.js.api import (DiscardPolicy, ExternalStream, StreamConfig,
                         StreamSource)


async def test(cloud: nats.aio.client.Client, leaf: nats.aio.client.Client):
    js_cloud = cloud.jetstream()
    js_leaf = leaf.jetstream()

    bucket_name = "test_subscriptions"

    obj_cloud = await js_cloud.create_object_store(bucket_name)

    await js_leaf.add_stream(
        StreamConfig(
            name=f"OBJ_{bucket_name}",
            discard=DiscardPolicy.NEW,
            max_msgs_per_subject=1,
            mirror=StreamSource(
                name=f"OBJ_{bucket_name}",
                external=ExternalStream(api="$JS.cloud.API", deliver=f"_INBOX.leaf-a.{bucket_name}"),
            ),
            deny_delete=True,
            allow_rollup_hdrs=True,
        )
    )
    obj_leaf = await js_leaf.object_store(bucket_name)

    await obj_cloud.put(name="key", data=b"data")

    # Throws an error when trying to obtain the stream name
    entry = await obj_leaf.get("key")

we call this function with appropriately initialized functions.

Given the capability you are leveraging, describe your expectation?

I get the entry I just wrote to the cloud object store from the leaf object store. This also the official way to sync the Key Value store (and apart from this issue also works for the object store): nats-io/nats-server#3557

Given the expectation, what is the defect you are observing?

This will generate an error roughly like this:

  File "/tests/test_object_store.py", line 58, in test_object_store_from_cloud_to_leaf
    entry = await object_store_leaf.get("key")
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/site-packages/nats/js/object_store.py", line 202, in get
    sub = await self._js.subscribe(
          ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/site-packages/nats/js/client.py", line 207, in subscribe
    stream = await self._jsm.find_stream_name_by_subject(subject)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/lib/python3.11/site-packages/nats/js/manager.py", line 67, in find_stream_name_by_subject
    return info['streams'][0]

This is the case because the subscribe in object_store.py. That file calls self._js.subscribe without a stream. The stream OBJ_test_subscriptions does not have subjects, but subscribing to it on the subject $O.test_subscriptions.> works nonetheless, since it mirrors another stream.

This should be fixable by just providing the stream (which the object store knows) to the subscribe call. This is already done in all calls that use get_last_msg instead of the raw subscribe. The Key Value store also provides the stream name to calls, except for the KeyValueWatcher, where the same problem occurs.

Not adding the stream argument is also inefficient, since it causes unnecessary calls to the NATS API.