diff --git a/README.md b/README.md index 2faa082..badd3e0 100644 --- a/README.md +++ b/README.md @@ -96,8 +96,9 @@ With `keyPrefix: 'myservice'`: | Key | Type | Owner | Purpose | |---|---|---|---| | `myservice:members` | set | pod | the set of memberIds known to be live | -| `myservice:member:` | hash with `address`, `load` | pod | live pod registration and load metric, TTL refreshed by heartbeat | +| `myservice:member:` | hash with `address`, `load` and, while draining, `draining` | pod | live pod registration and load metric, TTL refreshed by heartbeat | | `myservice:destination:` | set of memberIds | coordinator + pod | pods currently serving this destination | +| `myservice:owned:` | set of destination ids | coordinator + pod | the destinations of a pod, with the `memberIndex` option | | `myservice:lock:` | hash with `podId`, `destinationId`, metadata | pod | lockId routing for transaction-bound calls | ## Pod side: `Member` @@ -132,6 +133,10 @@ await member.unregisterLock(lockId) const peers = await member.listPeerLoad() // Graceful shutdown: +// Before stopping, stop receiving destinations while the ones held are moved or finished: the registry never binds a +// draining pod, nor reassigns orphans to it, and listPeerLoad() reports the flag +await member.setDraining(true) + clearInterval(heartbeat) await member.deregister() await member.close() @@ -169,14 +174,21 @@ if (lockRouting) { } // Other primitives: -await registry.listLiveMembers() // [{ memberId, address, load }, ...] +await registry.listLiveMembers() // [{ memberId, address, load, draining }, ...] await registry.pickMember({ destinationId: destId }) -await registry.addPodToDestination(destId, memberId) +await registry.addPodToDestination(destId, memberId) // false when the pod is draining: nothing is bound await registry.hasBinding(destId) await registry.deregisterDestination(destId) +await registry.listOwned(memberId) // the destinations of a pod, with the member index await registry.close() ``` +### Member index + +With `memberIndex: true`, on the `Registry` and on every `Member`, each binding also records the destination in the set of the pod, `myservice:owned:`, in the same script or transaction: claims and reassignments add it to the new pod and remove it from the pods unbound, pruning a dead pod removes it, `addPodToDestination`, `deregisterDestination`, `member.addToDestination()` and `member.removeFromDestination()` do the same. `registry.listOwned(memberId)` and `member.listOwned()` read the set, so that the destinations of a pod are listed without scanning them all, i.e. to move them before the pod stops. + +The index is exact only if every writer of the destinations maintains it, this library and any script of the application alike: enable it everywhere at once, and fill the sets of the destinations bound before, with a scan, before relying on it. + ## Resolution and failover `resolveDestination` reads the destination's pod set, filters by liveness, and applies the allocation strategy. The four cases: @@ -191,6 +203,8 @@ await registry.close() All writes use `SADD` / `SREM` (atomic). Concurrent first-touch by two coordinators can produce a destination with two pods from the start. That's a valid steady state, not a corrupted one. +A pod which called `member.setDraining(true)` is never picked, and the binding scripts refuse it even when it was picked from a list read before it started draining: `resolveDestination` then picks another pod, `addPodToDestination` returns `false`, and `pickAndRegister` answers 503 (`refusedMessage`, result `refused`), so that the caller retries on another pod. The flag is written again with the record of the pod when it expired. The destinations the pod already holds keep resolving to it until it deregisters or moves them. + ## Allocation strategies Pluggable. Built-in: `round-robin` (default), `least-loaded`, `random`. Custom strategies implement: diff --git a/src/helpers/pick-and-register.ts b/src/helpers/pick-and-register.ts index 5bb23ee..b09b66c 100644 --- a/src/helpers/pick-and-register.ts +++ b/src/helpers/pick-and-register.ts @@ -4,12 +4,14 @@ import type { FastifyReplyFromHooks } from '@fastify/reply-from' import '@fastify/reply-from' import type { Registry } from '../registry.ts' -export type PickAndRegisterResult = 'spawned' | 'unavailable' | 'upstream_error' +export type PickAndRegisterResult = 'spawned' | 'unavailable' | 'upstream_error' | 'refused' export interface PickAndRegisterOptions { registerIdFrom: (responseBody: any) => string expectedStatus?: number unavailableMessage?: string + /** The message of the 503 sent when the pod started draining after it was picked, so the binding was refused. */ + refusedMessage?: string onResult?: (result: PickAndRegisterResult) => void } @@ -21,6 +23,7 @@ export function pickAndRegister ( registerIdFrom, expectedStatus = 201, unavailableMessage = 'No pods available', + refusedMessage = 'The pod started draining, retry', onResult } = opts @@ -48,7 +51,15 @@ export function pickAndRegister ( if (res.statusCode === expectedStatus) { const id = registerIdFrom(body) registry.addPodToDestination(id, member.memberId).then( - () => { + (bound) => { + // The pod started draining between the pick and the binding: the resource it created is not routable, + // the caller retries on another pod + if (!bound) { + onResult?.('refused') + replyOut.code(503).send({ error: refusedMessage }) + return + } + onResult?.('spawned') replyOut.send(body) }, diff --git a/src/member.ts b/src/member.ts index 2ed51a6..d057799 100644 --- a/src/member.ts +++ b/src/member.ts @@ -12,6 +12,8 @@ export interface MemberOptions { keyPrefix?: string ttl?: number getLoad?: () => number + /** Maintains the set of the destinations of this member, see the `memberIndex` option of `Registry`. */ + memberIndex?: boolean } export class Member { @@ -22,6 +24,8 @@ export class Member { #keyPrefix: string #ttl: number #getLoad: () => number + #memberIndex: boolean + #draining: boolean constructor (opts: MemberOptions) { const { redis, owned } = connect(opts.redis, opts.redisOptions) @@ -32,6 +36,8 @@ export class Member { this.#keyPrefix = opts.keyPrefix ?? 'coordinator' this.#ttl = opts.ttl ?? 30 this.#getLoad = opts.getLoad ?? (() => 0) + this.#memberIndex = opts.memberIndex ?? false + this.#draining = false } get memberId (): string { @@ -54,6 +60,10 @@ export class Member { return `${this.#keyPrefix}:destination:${destId}` } + #ownedKey (): string { + return `${this.#keyPrefix}:owned:${this.#memberId}` + } + #lockKey (lockId: string): string { return `${this.#keyPrefix}:lock:${lockId}` } @@ -71,9 +81,11 @@ export class Member { async #writeRecord (): Promise { const pipeline = this.#redis.pipeline() pipeline.sadd(this.#membersKey(), this.#memberId) + // A record written again after it expired keeps the member draining pipeline.hset(this.#memberKey(), { address: this.#address, - load: String(this.#getLoad()) + load: String(this.#getLoad()), + ...(this.#draining ? { draining: '1' } : {}) }) pipeline.expire(this.#memberKey(), this.#ttl) await pipeline.exec() @@ -86,12 +98,42 @@ export class Member { await pipeline.exec() } + // A draining member keeps the destinations it holds, but is bound to no new one, nor reassigned any, until it + // resumes: the registry checks the flag atomically when it binds. The heartbeats keep the flag, and write it again + // with a record which expired; it goes when the member resumes or deregisters. + async setDraining (draining: boolean): Promise { + this.#draining = draining + + if (draining) { + await this.#redis.hset(this.#memberKey(), 'draining', '1') + } else { + await this.#redis.hdel(this.#memberKey(), 'draining') + } + } + + get draining (): boolean { + return this.#draining + } + async addToDestination (destinationId: string): Promise { - await this.#redis.sadd(this.#destinationKey(destinationId), this.#memberId) + if (this.#memberIndex) { + await this.#redis.multi().sadd(this.#destinationKey(destinationId), this.#memberId).sadd(this.#ownedKey(), destinationId).exec() + } else { + await this.#redis.sadd(this.#destinationKey(destinationId), this.#memberId) + } } async removeFromDestination (destinationId: string): Promise { - await this.#redis.srem(this.#destinationKey(destinationId), this.#memberId) + if (this.#memberIndex) { + await this.#redis.multi().srem(this.#destinationKey(destinationId), this.#memberId).srem(this.#ownedKey(), destinationId).exec() + } else { + await this.#redis.srem(this.#destinationKey(destinationId), this.#memberId) + } + } + + // The destinations of this member, from the member index + async listOwned (): Promise { + return this.#redis.smembers(this.#ownedKey()) } async registerLock ( @@ -116,7 +158,7 @@ export class Member { const pipeline = this.#redis.pipeline() for (const id of memberIds) { - pipeline.hmget(this.#memberKey(id), 'address', 'load') + pipeline.hmget(this.#memberKey(id), 'address', 'load', 'draining') } const results = await pipeline.exec() if (!results) return [] @@ -128,7 +170,7 @@ export class Member { const address = fields[0] if (!address) continue const load = parseInt(fields[1] ?? '0', 10) || 0 - live.push({ memberId: memberIds[i], address, load }) + live.push({ memberId: memberIds[i], address, load, draining: fields[2] === '1' }) } return live } diff --git a/src/registry.ts b/src/registry.ts index d90cf91..d6a30e0 100644 --- a/src/registry.ts +++ b/src/registry.ts @@ -12,6 +12,12 @@ export interface RegistryOptions { strategy?: 'round-robin' | 'least-loaded' | 'random' | AllocationStrategy requestTimeout?: number cache?: CacheOptions | false + /** + * Maintains a set of the destinations bound to each member, `:owned:`, atomically with every + * binding made by this registry, so that the destinations of a member are listed without scanning them all. Every + * writer of the destinations must maintain it, see the README. + */ + memberIndex?: boolean } export interface ResolveResult { @@ -31,7 +37,9 @@ export interface ResolveLockResult { // members are pruned when at least one is live. Without a candidate (ARGV[2]) it only reports how many members the // set holds; with one, it binds the candidate when the set is empty (ARGV[3] = '1', claim) or when every member is dead // (ARGV[4] = '1', reassign). Concurrent resolutions therefore agree: a caller losing the race gets the members bound by -// the winner instead of adding its own candidate. +// the winner instead of adding its own candidate. A candidate which is draining is refused: the caller picks another. +// With a member index (ARGV[5] is its key prefix, ARGV[6] the destination), the sets of the members bound and unbound +// are updated in the same script. const resolveScript = ` local members = redis.call('SMEMBERS', KEYS[1]) local live = {} @@ -52,6 +60,12 @@ const resolveScript = ` if #live > 0 then if #dead > 0 then redis.call('SREM', KEYS[1], unpack(dead)) + + if ARGV[5] ~= '' then + for _, member in ipairs(dead) do + redis.call('SREM', ARGV[5] .. member, ARGV[6]) + end + end end return { 'live', live } @@ -63,18 +77,63 @@ const resolveScript = ` return { 'empty', #members } end - local address = redis.call('HGET', ARGV[1] .. ARGV[2], 'address') + local fields = redis.call('HMGET', ARGV[1] .. ARGV[2], 'address', 'draining') - if not address then + if not fields[1] then return { 'stale' } end + -- A draining member receives no new destination: checked here, atomically with the binding, since the candidate + -- was picked from a list read before, when the member might not have been draining yet + if fields[2] == '1' then + return { 'draining' } + end + + local address = fields[1] redis.call('DEL', KEYS[1]) redis.call('SADD', KEYS[1], ARGV[2]) + if ARGV[5] ~= '' then + for _, member in ipairs(members) do + redis.call('SREM', ARGV[5] .. member, ARGV[6]) + end + + redis.call('SADD', ARGV[5] .. ARGV[2], ARGV[6]) + end + return { 'bound', { ARGV[2], address }, dead } ` +// Binds a member to a destination explicitly (addPodToDestination), unless the member is draining: refused atomically, +// as by the resolve script, since the member can start draining after it was picked. With a member index (ARGV[3], +// ARGV[4]), the destination is recorded in the set of the member. +const bindScript = ` + if redis.call('HGET', ARGV[1], 'draining') == '1' then + return 0 + end + + redis.call('SADD', KEYS[1], ARGV[2]) + + if ARGV[3] ~= '' then + redis.call('SADD', ARGV[3] .. ARGV[2], ARGV[4]) + end + + return 1 +` + +// Deletes a destination and, with a member index (ARGV[1], ARGV[2]), removes it from the sets of its members, in one +// script: a member bound between a read and the deletion would otherwise keep a stale entry +const deregisterScript = ` + if ARGV[1] ~= '' then + for _, member in ipairs(redis.call('SMEMBERS', KEYS[1])) do + redis.call('SREM', ARGV[1] .. member, ARGV[2]) + end + end + + redis.call('DEL', KEYS[1]) + return 1 +` + // A candidate can expire between the pick and the script, in which case another one is picked const maxBindAttempts = 3 @@ -82,10 +141,13 @@ type ResolveReply = | ['live', string[]] | ['empty', number] | ['stale'] + | ['draining'] | ['bound', [string, string], string[]] interface ResolveCommand { - coordinatorResolve (key: string, memberPrefix: string, candidate: string, claim: string, reassign: string): Promise + coordinatorResolve (key: string, memberPrefix: string, candidate: string, claim: string, reassign: string, ownedPrefix: string, destinationId: string): Promise + coordinatorBind (key: string, memberKey: string, memberId: string, ownedPrefix: string, destinationId: string): Promise + coordinatorDeregister (key: string, ownedPrefix: string, destinationId: string): Promise } export class Registry { @@ -94,11 +156,14 @@ export class Registry { #keyPrefix: string #strategy: AllocationStrategy #cache: TTLCache | null + #memberIndex: boolean readonly requestTimeout: number | undefined constructor (opts: RegistryOptions) { const { redis, owned } = connect(opts.redis, opts.redisOptions) redis.defineCommand('coordinatorResolve', { numberOfKeys: 1, lua: resolveScript }) + redis.defineCommand('coordinatorBind', { numberOfKeys: 1, lua: bindScript }) + redis.defineCommand('coordinatorDeregister', { numberOfKeys: 1, lua: deregisterScript }) this.#redis = redis as Redis & ResolveCommand this.#ownsRedis = owned this.#keyPrefix = opts.keyPrefix ?? 'coordinator' @@ -113,6 +178,7 @@ export class Registry { ? null : new TTLCache(opts.cache) this.requestTimeout = opts.requestTimeout + this.#memberIndex = opts.memberIndex ?? false } #membersKey (): string { @@ -131,13 +197,22 @@ export class Registry { return `${this.#keyPrefix}:lock:${lockId}` } + // The prefix of the sets of the member index, or '' when the index is not maintained + #ownedPrefix (): string { + return this.#memberIndex ? `${this.#keyPrefix}:owned:` : '' + } + + #ownedKey (memberId: string): string { + return `${this.#keyPrefix}:owned:${memberId}` + } + async listLiveMembers (): Promise { const memberIds = await this.#redis.smembers(this.#membersKey()) if (memberIds.length === 0) return [] const pipeline = this.#redis.pipeline() for (const id of memberIds) { - pipeline.hmget(this.#memberKey(id), 'address', 'load') + pipeline.hmget(this.#memberKey(id), 'address', 'load', 'draining') } const results = await pipeline.exec() if (!results) return [] @@ -149,14 +224,17 @@ export class Registry { const address = fields[0] if (!address) continue const load = parseInt(fields[1] ?? '0', 10) || 0 - live.push({ memberId: memberIds[i], address, load }) + live.push({ memberId: memberIds[i], address, load, draining: fields[2] === '1' }) } return live } - async pickMember (ctx: { destinationId?: string } = {}): Promise { - const live = await this.listLiveMembers() - return this.#strategy.pick(live, ctx) + // Picks a live member for a new destination with the strategy, among the members which are not draining, nor + // excluded: the members which refused the binding meanwhile + async pickMember (ctx: { destinationId?: string, exclude?: Iterable } = {}): Promise { + const excluded = new Set(ctx.exclude ?? []) + const live = (await this.listLiveMembers()).filter(member => !member.draining && !excluded.has(member.memberId)) + return this.#strategy.pick(live, { destinationId: ctx.destinationId }) } async resolveDestination ( @@ -181,7 +259,10 @@ export class Registry { const claim = opts.claimOnMiss ? '1' : '0' const reassign = opts.reassignOrphans ? '1' : '0' - let reply = await this.#redis.coordinatorResolve(key, memberPrefix, '', claim, reassign) + const ownedPrefix = this.#ownedPrefix() + let reply = await this.#redis.coordinatorResolve(key, memberPrefix, '', claim, reassign, ownedPrefix, destinationId) + const refused: string[] = [] + let picked: MemberInfo | null = null for (let attempt = 0; attempt < maxBindAttempts; attempt++) { if (reply[0] === 'live') { @@ -200,10 +281,15 @@ export class Registry { if (!wanted) return null } - const picked = await this.pickMember({ destinationId }) + // A candidate which started draining after it was picked is not picked again + if (reply[0] === 'draining' && picked) { + refused.push(picked.memberId) + } + + picked = await this.pickMember({ destinationId, exclude: refused }) if (!picked) return null - reply = await this.#redis.coordinatorResolve(key, memberPrefix, picked.memberId, claim, reassign) + reply = await this.#redis.coordinatorResolve(key, memberPrefix, picked.memberId, claim, reassign, ownedPrefix, destinationId) } return reply[0] === 'live' ? this.#pickLive(destinationId, reply[1]) : null @@ -221,9 +307,17 @@ export class Registry { return { address: picked.address, memberId: picked.memberId, reassigned: false } } - async addPodToDestination (destinationId: string, memberId: string): Promise { - await this.#redis.sadd(this.#destinationKey(destinationId), memberId) + // Binds a member to a destination. Returns false when the member is draining, in which case nothing is bound: the + // member can start draining after it was picked. + async addPodToDestination (destinationId: string, memberId: string): Promise { + const bound = await this.#redis.coordinatorBind(this.#destinationKey(destinationId), this.#memberKey(memberId), memberId, this.#ownedPrefix(), destinationId) if (this.#cache) this.#cache.delete(destinationId) + return bound === 1 + } + + // The destinations bound to a member, from the member index + async listOwned (memberId: string): Promise { + return this.#redis.smembers(this.#ownedKey(memberId)) } async hasBinding (destinationId: string): Promise { @@ -244,7 +338,7 @@ export class Registry { } async deregisterDestination (destinationId: string): Promise { - await this.#redis.del(this.#destinationKey(destinationId)) + await this.#redis.coordinatorDeregister(this.#destinationKey(destinationId), this.#ownedPrefix(), destinationId) if (this.#cache) this.#cache.delete(destinationId) } diff --git a/src/strategies.ts b/src/strategies.ts index 47089d7..31d801e 100644 --- a/src/strategies.ts +++ b/src/strategies.ts @@ -2,6 +2,8 @@ export interface MemberInfo { memberId: string address: string load: number + // A draining member keeps serving the destinations it holds, but receives no new ones + draining?: boolean } export interface PickContext { diff --git a/test/helpers.test.ts b/test/helpers.test.ts index 5e04fe6..62a77e9 100644 --- a/test/helpers.test.ts +++ b/test/helpers.test.ts @@ -1,4 +1,4 @@ -import { strictEqual, ok } from 'node:assert' +import { deepStrictEqual, strictEqual, ok } from 'node:assert' import { randomBytes } from 'node:crypto' import test from 'node:test' import Fastify from 'fastify' @@ -142,6 +142,45 @@ test('Coordinator helpers', async (t) => { ok(set[0] === memberId1 || set[0] === memberId2) }) + await t.test('pickAndRegister: refuses the binding when the pod started draining after it was picked', async () => { + // A pod which starts draining while it creates the resource + let created = 0 + const draining = Fastify() + draining.post('/resources', async (_req, reply) => { + await redis.hset(memberKey('draining-pod'), 'draining', '1') + created++ + return reply.code(201).send({ resourceId: 'r-draining' }) + }) + await draining.listen({ port: 0, host: '127.0.0.1' }) + await makeLivePod(redis, 'draining-pod', `http://127.0.0.1:${(draining.server.address() as any).port}`) + + const results: string[] = [] + const picky = new Registry({ + redis: REDIS_URL, + keyPrefix: PREFIX, + cache: false, + strategy: { pick: (candidates) => candidates.find(candidate => candidate.memberId === 'draining-pod') ?? null } + }) + const app = Fastify() + await app.register(replyFrom) + app.post('/resources', pickAndRegister(picky, { registerIdFrom: (r: any) => r.resourceId, onResult: result => results.push(result) })) + + try { + const res = await app.inject({ method: 'POST', url: '/resources' }) + strictEqual(res.statusCode, 503) + ok((res.json() as any).error.includes('draining')) + deepStrictEqual(results, ['refused']) + strictEqual(created, 1) + strictEqual(await redis.exists(destinationKey('r-draining')), 0) + } finally { + await app.close() + await picky.close() + await draining.close() + await redis.srem(membersKey(), 'draining-pod') + await redis.del(memberKey('draining-pod')) + } + }) + await t.test('pickAndRegister: returns 503 when no pods are available', async () => { const isolatedPrefix = `${PREFIX}-empty-${randomBytes(2).toString('hex')}` const emptyRegistry = new Registry({ redis: REDIS_URL, keyPrefix: isolatedPrefix, cache: false }) diff --git a/test/member.test.ts b/test/member.test.ts index c29d9f8..1748e83 100644 --- a/test/member.test.ts +++ b/test/member.test.ts @@ -36,6 +36,46 @@ test('Member', async (t) => { await sharedRedis.quit() }) + await t.test('setDraining marks the member in its record, across heartbeats, until it resumes', async () => { + await member.register() + await member.setDraining(true) + strictEqual(await sharedRedis.hget(memberKey(memberId), 'draining'), '1') + await member.heartbeat() + strictEqual(await sharedRedis.hget(memberKey(memberId), 'draining'), '1') + deepStrictEqual((await member.listPeerLoad()).map(peer => [peer.memberId, peer.draining]), [[memberId, true]]) + + // A record which expired is written again with the flag + await sharedRedis.del(memberKey(memberId)) + await member.heartbeat() + strictEqual(await sharedRedis.hget(memberKey(memberId), 'draining'), '1') + strictEqual(member.draining, true) + + await member.setDraining(false) + strictEqual(member.draining, false) + strictEqual(await sharedRedis.hget(memberKey(memberId), 'draining'), null) + deepStrictEqual((await member.listPeerLoad()).map(peer => [peer.memberId, peer.draining]), [[memberId, false]]) + }) + + await t.test('a member with the index records the destinations it joins and leaves', async () => { + const indexed = new Member({ redis: REDIS_URL, memberId: 'indexed-member', address, keyPrefix: PREFIX, memberIndex: true }) + + try { + await indexed.addToDestination('dest-owned-1') + await indexed.addToDestination('dest-owned-2') + deepStrictEqual((await indexed.listOwned()).sort(), ['dest-owned-1', 'dest-owned-2']) + await indexed.removeFromDestination('dest-owned-1') + deepStrictEqual(await indexed.listOwned(), ['dest-owned-2']) + deepStrictEqual(await sharedRedis.smembers(destinationKey('dest-owned-2')), ['indexed-member']) + + // Without the index, nothing is recorded + await member.addToDestination('dest-plain') + deepStrictEqual(await member.listOwned(), []) + } finally { + await indexed.close() + await sharedRedis.del(destinationKey('dest-owned-1'), destinationKey('dest-owned-2'), destinationKey('dest-plain'), `${PREFIX}:owned:indexed-member`) + } + }) + await t.test('register adds member to set and writes hash with TTL', async () => { connections = 7 await member.register() diff --git a/test/registry.test.ts b/test/registry.test.ts index 69b5ce8..400a73d 100644 --- a/test/registry.test.ts +++ b/test/registry.test.ts @@ -188,6 +188,119 @@ test('Registry', async (t) => { } }) + await t.test('resolveDestination never binds a draining member, even one picked before it started draining', async () => { + await sharedRedis.hset(memberKey(m1), 'draining', '1') + + try { + // Draining members are not candidates + for (let i = 0; i < 5; i++) { + const resolved = await registry.resolveDestination(`inst-draining-${i}`, { claimOnMiss: true }) + ok(resolved) + strictEqual(resolved.memberId, m2) + } + + deepStrictEqual((await registry.listLiveMembers()).map(member => [member.memberId, member.draining]).sort(), [[m1, true], [m2, false]]) + + // A strategy which picked the member before it started draining: the binding is refused, and another member picked + let picks = 0 + const picky = new Registry({ + redis: REDIS_URL, + keyPrefix: PREFIX, + cache: false, + strategy: { + pick (candidates) { + if (picks++ === 0) return { memberId: m1, address: m1Address, load: 0 } + return candidates[0] ?? null + } + } + }) + + try { + const resolved = await picky.resolveDestination('inst-draining-race', { claimOnMiss: true }) + ok(resolved) + strictEqual(resolved.memberId, m2) + strictEqual(picks, 2) + deepStrictEqual(await sharedRedis.smembers(destinationKey('inst-draining-race')), [m2]) + + // Orphans are not reassigned to it either + await sharedRedis.sadd(destinationKey('inst-draining-orphan'), 'dead-pod') + picks = 0 + const reassigned = await picky.resolveDestination('inst-draining-orphan', { reassignOrphans: true }) + ok(reassigned) + strictEqual(reassigned.memberId, m2) + } finally { + await picky.close() + } + + // The destinations it holds are still served by it + await sharedRedis.sadd(destinationKey('inst-draining-held'), m1) + strictEqual((await registry.resolveDestination('inst-draining-held'))?.memberId, m1) + + // Without another member, nothing is bound + await sharedRedis.hset(memberKey(m2), 'draining', '1') + strictEqual(await registry.resolveDestination('inst-draining-none', { claimOnMiss: true }), null) + + // An explicit binding is refused too + strictEqual(await registry.addPodToDestination('inst-draining-explicit', m1), false) + deepStrictEqual(await sharedRedis.exists(destinationKey('inst-draining-explicit')), 0) + } finally { + await sharedRedis.hdel(memberKey(m1), 'draining') + await sharedRedis.hdel(memberKey(m2), 'draining') + for (const id of ['inst-draining-race', 'inst-draining-orphan', 'inst-draining-held', ...Array.from({ length: 5 }, (_, i) => `inst-draining-${i}`)]) { + await sharedRedis.del(destinationKey(id)) + } + } + }) + + await t.test('the member index follows the bindings of the registry which maintains it', async () => { + const indexed = new Registry({ redis: REDIS_URL, keyPrefix: PREFIX, cache: false, memberIndex: true, strategy: 'round-robin' }) + const ownedKey = (id: string): string => `${PREFIX}:owned:${id}` + + try { + // Claims add the destination to the set of the member bound + const first = await indexed.resolveDestination('inst-owned-1', { claimOnMiss: true }) + const second = await indexed.resolveDestination('inst-owned-2', { claimOnMiss: true }) + ok(first && second) + deepStrictEqual((await indexed.listOwned(first.memberId)).includes('inst-owned-1'), true) + deepStrictEqual((await indexed.listOwned(second.memberId)).includes('inst-owned-2'), true) + + // A reassignment moves the destination from the dead member to the new one + await sharedRedis.sadd(destinationKey('inst-owned-orphan'), 'dead-pod') + await sharedRedis.sadd(ownedKey('dead-pod'), 'inst-owned-orphan', 'inst-owned-pruned') + const reassigned = await indexed.resolveDestination('inst-owned-orphan', { reassignOrphans: true }) + ok(reassigned) + deepStrictEqual(await sharedRedis.sismember(ownedKey('dead-pod'), 'inst-owned-orphan'), 0) + deepStrictEqual(await sharedRedis.sismember(ownedKey(reassigned.memberId), 'inst-owned-orphan'), 1) + + // Pruning a dead member from a destination with live ones removes it from its set + await sharedRedis.sadd(destinationKey('inst-owned-pruned'), m1, 'dead-pod') + await sharedRedis.sadd(ownedKey(m1), 'inst-owned-pruned') + ok(await indexed.resolveDestination('inst-owned-pruned')) + deepStrictEqual(await sharedRedis.sismember(ownedKey('dead-pod'), 'inst-owned-pruned'), 0) + deepStrictEqual(await sharedRedis.sismember(ownedKey(m1), 'inst-owned-pruned'), 1) + + // Explicit bindings and deregistrations maintain it too + strictEqual(await indexed.addPodToDestination('inst-owned-added', m2), true) + deepStrictEqual(await sharedRedis.sismember(ownedKey(m2), 'inst-owned-added'), 1) + await indexed.deregisterDestination('inst-owned-added') + deepStrictEqual(await sharedRedis.sismember(ownedKey(m2), 'inst-owned-added'), 0) + deepStrictEqual(await sharedRedis.exists(destinationKey('inst-owned-added')), 0) + + // A registry without the index writes none of it + const plain = await registry.resolveDestination('inst-owned-plain', { claimOnMiss: true }) + ok(plain) + deepStrictEqual(await sharedRedis.sismember(ownedKey(plain.memberId), 'inst-owned-plain'), 0) + } finally { + await indexed.close() + for (const id of ['inst-owned-1', 'inst-owned-2', 'inst-owned-orphan', 'inst-owned-pruned', 'inst-owned-added', 'inst-owned-plain']) { + await sharedRedis.del(destinationKey(id)) + } + for (const id of [m1, m2, 'dead-pod']) { + await sharedRedis.del(ownedKey(id)) + } + } + }) + await t.test('resolveDestination with multi-pod set picks one live pod, cleans dead members in background', async () => { await sharedRedis.sadd(destinationKey('inst-multi'), m1, 'dead-pod', m2) const result = await registry.resolveDestination('inst-multi')