Distributed locks with Nats?

[kevinhikaruevans]

I was wondering if there's an established way of doing distributed locks with Nats.

Currently, I'm using kv and this seems to work, but I'm not too sure about timing/edge cases. Lock order/sequence isn't needed for my application and I presume to correctly do this in FIFO, we'd need to use a queue or something.

To acquire, I'm creating a key-value pair in a bucket, then deleting it on release. If there's an existing lock, it'll watch til the lock is deleted before trying again. The _locked attribute here is mostly to prevent double releasing/acquiring. On startup, I'm deleting & recreating the kv bucket.

My class looks like:

class NatsAsyncLock:
    def __init__(self, kv: KeyValue, name: str):
        self.kv = kv
        self.name = name
        self.owner_id = str(uuid.uuid4())
        self._locked = False
        self.logger = logging.getLogger(f"NatsAsyncLock")


    async def acquire(self):
        self.logger.info("acquiring...")

        while True:
            try:
                await self.kv.create(self.name, self.owner_id.encode())
                
                self.logger.info("acquired!")
                self._locked = True
                return True
            except Exception:
                watcher = await self.kv.watch(self.name)
                
                while True:
                    self.logger.info("could not acquire (locked?)")
                    event = await watcher.updates()

                    if event is None:
                        self.logger.info("waiting...")
                        continue
                    
                    if event.operation == KV_DEL:
                        self.logger.info("was deleted! ^_^")
                        break

    async def release(self):
        if not self._locked:
            return
        try:
            self.logger.info("releasing...")
            entry = await self.kv.get(self.name)
            if entry.value is not None and entry.value.decode() == self.owner_id:
                await self.kv.delete(self.name)
        finally:
            self.logger.info("released!")
            self._locked = False

    async def __aenter__(self):
        await self.acquire()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.release()

[kevinhikaruevans]

Just realized this was more/less described in the docs here: https://docs.nats.io/nats-concepts/jetstream/key-value-store/kv_walkthrough#create-aka-exclusive-locking