mongo/me.ahoo.wow.mongo.subscription/MongoLeaseStore

MongoLeaseStore

class MongoLeaseStore(client: MongoClient, databaseName: String, clock: Clock = Clock.systemUTC(), dispatcher: CoroutineDispatcher = VirtualThreads.dispatcher, ddl: Ddl = Ddl.AUTO, operationTimeout: Duration = MongoTimeouts.OPERATION) : LeaseStore

The LeaseStore on MongoDB: one document per (group, partition) in wow_lease and one per (group, node), with the node's internal address and the lease groups it serves, in wow_lease_member (one per service and node, its key field holding the service), each keyed by a unique index.

Every operation is one single-document write. An acquisition is a conditional upsert (see conditionalUpsert) whose filter admits only an expired lease or the acquirer's own, and whose pipeline update keeps the epoch when the acquirer still holds the lease live and adds one otherwise. MongoDB re-evaluates the filter after a write conflict, so concurrent acquisitions have exactly one winner. A release keeps the document with expiry 0, so the next acquisition continues its epoch.

Times come from clock, so they are the calling node's: keep the TTL well above the nodes' clock skew. Calls run on dispatcher, virtual threads by default.

Constructors

MongoLeaseStore

constructor(client: MongoClient, databaseName: String, clock: Clock = Clock.systemUTC(), dispatcher: CoroutineDispatcher = VirtualThreads.dispatcher, ddl: Ddl = Ddl.AUTO, operationTimeout: Duration = MongoTimeouts.OPERATION)

Functions

acquire

open suspend override fun acquire(group: String, partition: Int, owner: String, ttl: Duration): Lease?

createSchema

fun createSchema()

Creates the key indexes when missing.

heartbeat

open suspend override fun heartbeat(service: String, node: String, address: String, ttl: Duration, groups: Set<String>): Member

leave

open suspend override fun leave(service: String, node: String)

members

open suspend override fun members(service: String): List<Member>

owners

open suspend override fun owners(group: String): List<Lease>

release

open suspend override fun release(group: String, partition: Int, owner: String): Boolean

renew

open suspend override fun renew(group: String, partitions: Collection<Int>, owner: String, ttl: Duration): List<Lease>

Two round trips for every partition: the live leases owner holds are extended to one expiry, then the leases at that expiry are read back. A node renews its leases one round at a time, so no other call of owner sets the same expiry in between.