NATS KV for Caching (timeout errors)

[theobouwman]

Our caching solution with NATS works really great! We use a hosted NATS service and out FastAPI GCP Cloud Run instances (minimum 4) connect to it to write and read JetStream KV items.

But, after around 1 day we get errors in Sentry like:

• [Errno 104] Connection reset by peer
• nats: timeout
• Error getting key value: nats: timeout
• ConnectionDoesNotExistError

And then after a few minutes it works again like a champ.

Looking at the NatsService initialisation we just do this on FastAPI instance startup once. so with out current setup of 4 instances this is just 4 connections.

Were can we look to resolve or debug this issue? Has anyone else used NATS as KV also? Or are we implementing it in a totally wrong way?

Thanks.

Here is the code we use:

Our NATS Service:

from faststream.nats import NatsBroker
from nats.js.api import KeyValueConfig
from nats.js.errors import KeyNotFoundError

class NATSService:
    def __init__(self, broker: NatsBroker):
        self.broker = broker


    async def init(self):
        """
        create key value stores for each bucket
        """
        try:
            tasks = []
            for bucket in NATSBuckets:
                tasks.append(asyncio.create_task(self.broker.stream.create_key_value(KeyValueConfig(
                    bucket=bucket.value,
                    ttl=60*60*24*7 # 1 week
                ))))

            await asyncio.gather(*tasks)
        except Exception as e:
            logging.error(f"Error creating key value stores: {e}")



    def get_cache_key_organisation_event(self, organisation_id: str, ending_after: AwareDatetime, category: str = None):
        """
        key for caching organisation events
        """
        key = f"organisation.{organisation_id}.events.{ending_after.date()}"

        if category:
            key = f"{key}.{category}"

        return key


    async def publish(self, message: BaseModel, subject: str, stream: str | None = None):
        await self.broker.publish(message, subject, stream=stream)


    async def _key_value(self, bucket):
        return await self.broker.stream.key_value(bucket)


    async def get_key_value(self, bucket: str, key: str):
        try:
            result = await (await self._key_value(bucket)).get(key)
            return result
        except KeyNotFoundError as e:
            return None
        except Exception as e:
            logging.error(f"Error getting key value: {e}")
            sentry_sdk.capture_exception(e)
            return None



    async def set_key_value(self, bucket: str, key: str, value: bytes):
        try:
            await (await self._key_value(bucket)).put(key, value)
        except Exception as e:
            logging.error(f"Error setting key value: {e}")
            sentry_sdk.capture_exception(e)


    async def delete_key(self, bucket: str, key: str):
        try:
            await (await self._key_value(bucket)).delete(key)
        except Exception as e:
            logging.error(f"Error deleting key value: {e}")
            sentry_sdk.capture_exception(e)

The NATS Service gets initialised on instance start with Dishka DI https://github.com/reagento/dishka:

class NATSProvider(Provider):
    @provide(scope=Scope.APP)
    async def broker(self) -> NATSService:
        async def error_cb(e):
            logging.error(f"Error in NATS: {e}")
            sentry_sdk.capture_exception(e)


        _broker = NatsBroker(
            get_config().NATS_SERVER_URL,
            user_credentials="./nats-production-api.creds",
            error_cb=error_cb,
        )
        client = await _broker.connect()

        nats_service = NATSService(_broker)
        await nats_service.init()

        return nats_service

...


container = make_async_container(
    NATSProvider(),
    ...
)

And this is how we retrieve the cached items:

cached = await self.nats_service.get_key_value(NATSBuckets.ORGANISATION_EVENTS.value, "cache_key_xxx")

[AbstractiveNord]

• First of all, network is never-guaranteed thing. Time-outs is OK until it's too much. How often it happens? What's network quality?
  Are you using any network retry library? What's packet losses in your setup?

• Second, you are have problem at

  async def _key_value(self, bucket):
       return await self.broker.stream.key_value(bucket)
  

  The meaning is: "Query NATS Jetstream metadata about stream". Cache that data, don't query that every single time.

• How much streams you have? They have identical TTL, just different names.

• If you have not so much data, you can bring an excellent Watch feature of NATS KV. You can store some values directly in RAM of every worker, and update them with incoming KV messages. Even if data is more than single worker machine RAM, you can use Watch on specific key or key part, because it's all subjects under the hood. Bring pieces on demand. Forget them too. That's much faster, because you eliminate a network round trip at all.