From 529e49323ba690258942fae2bb4dea1d821addca Mon Sep 17 00:00:00 2001 From: marcopiraccini Date: Sat, 14 Nov 2026 09:48:23 +0100 Subject: [PATCH 1/3] feat: refuse to bind destinations to draining members Signed-off-by: marcopiraccini --- README.md | 10 ++++++-- src/member.ts | 15 +++++++++-- src/registry.ts | 36 +++++++++++++++++++------- src/strategies.ts | 2 ++ test/member.test.ts | 12 +++++++++ test/registry.test.ts | 60 +++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 122 insertions(+), 13 deletions(-) diff --git a/README.md b/README.md index 2faa082..c7dff32 100644 --- a/README.md +++ b/README.md @@ -96,7 +96,7 @@ 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:lock:` | hash with `podId`, `destinationId`, metadata | pod | lockId routing for transaction-bound calls | @@ -132,6 +132,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,7 +173,7 @@ 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.hasBinding(destId) @@ -191,6 +195,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 script refuses it even when it was picked from a list read before it started draining: the coordinator then picks another pod. The destinations it 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/member.ts b/src/member.ts index 2ed51a6..1b233e2 100644 --- a/src/member.ts +++ b/src/member.ts @@ -86,6 +86,17 @@ 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 flag survives the heartbeats, and goes with + // the record when it expires or the member deregisters. + async setDraining (draining: boolean): Promise { + if (draining) { + await this.#redis.hset(this.#memberKey(), 'draining', '1') + } else { + await this.#redis.hdel(this.#memberKey(), 'draining') + } + } + async addToDestination (destinationId: string): Promise { await this.#redis.sadd(this.#destinationKey(destinationId), this.#memberId) } @@ -116,7 +127,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 +139,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..9b2d572 100644 --- a/src/registry.ts +++ b/src/registry.ts @@ -31,7 +31,7 @@ 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. const resolveScript = ` local members = redis.call('SMEMBERS', KEYS[1]) local live = {} @@ -63,12 +63,19 @@ 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]) @@ -82,6 +89,7 @@ type ResolveReply = | ['live', string[]] | ['empty', number] | ['stale'] + | ['draining'] | ['bound', [string, string], string[]] interface ResolveCommand { @@ -137,7 +145,7 @@ export class Registry { 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 +157,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 ( @@ -182,6 +193,8 @@ export class Registry { const reassign = opts.reassignOrphans ? '1' : '0' let reply = await this.#redis.coordinatorResolve(key, memberPrefix, '', claim, reassign) + const refused: string[] = [] + let picked: MemberInfo | null = null for (let attempt = 0; attempt < maxBindAttempts; attempt++) { if (reply[0] === 'live') { @@ -200,7 +213,12 @@ 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) 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/member.test.ts b/test/member.test.ts index c29d9f8..bc29aba 100644 --- a/test/member.test.ts +++ b/test/member.test.ts @@ -36,6 +36,18 @@ 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]]) + await member.setDraining(false) + strictEqual(await sharedRedis.hget(memberKey(memberId), 'draining'), null) + deepStrictEqual((await member.listPeerLoad()).map(peer => [peer.memberId, peer.draining]), [[memberId, false]]) + }) + 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..500f7d2 100644 --- a/test/registry.test.ts +++ b/test/registry.test.ts @@ -188,6 +188,66 @@ 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) + } 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('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') From a4e48aac9676c868cd9e7127287de7e070559a1b Mon Sep 17 00:00:00 2001 From: marcopiraccini Date: Sat, 14 Nov 2026 09:55:15 +0100 Subject: [PATCH 2/3] feat: index the destinations of each member Signed-off-by: marcopiraccini --- README.md | 8 ++++++ src/member.ts | 25 ++++++++++++++++-- src/registry.ts | 61 +++++++++++++++++++++++++++++++++++++++---- test/member.test.ts | 20 ++++++++++++++ test/registry.test.ts | 49 ++++++++++++++++++++++++++++++++++ 5 files changed, 156 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index c7dff32..b1d3499 100644 --- a/README.md +++ b/README.md @@ -98,6 +98,7 @@ With `keyPrefix: 'myservice'`: | `myservice:members` | set | pod | the set of memberIds known to be live | | `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` @@ -178,9 +179,16 @@ await registry.pickMember({ destinationId: destId }) await registry.addPodToDestination(destId, memberId) 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: diff --git a/src/member.ts b/src/member.ts index 1b233e2..275d952 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,7 @@ export class Member { #keyPrefix: string #ttl: number #getLoad: () => number + #memberIndex: boolean constructor (opts: MemberOptions) { const { redis, owned } = connect(opts.redis, opts.redisOptions) @@ -32,6 +35,7 @@ export class Member { this.#keyPrefix = opts.keyPrefix ?? 'coordinator' this.#ttl = opts.ttl ?? 30 this.#getLoad = opts.getLoad ?? (() => 0) + this.#memberIndex = opts.memberIndex ?? false } get memberId (): string { @@ -54,6 +58,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}` } @@ -98,11 +106,24 @@ export class Member { } 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 ( diff --git a/src/registry.ts b/src/registry.ts index 9b2d572..8dcd405 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 { @@ -32,6 +38,8 @@ export interface ResolveLockResult { // 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. 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 } @@ -79,6 +93,14 @@ const resolveScript = ` 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 } ` @@ -93,7 +115,7 @@ type ResolveReply = | ['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 } export class Registry { @@ -102,6 +124,7 @@ export class Registry { #keyPrefix: string #strategy: AllocationStrategy #cache: TTLCache | null + #memberIndex: boolean readonly requestTimeout: number | undefined constructor (opts: RegistryOptions) { @@ -121,6 +144,7 @@ export class Registry { ? null : new TTLCache(opts.cache) this.requestTimeout = opts.requestTimeout + this.#memberIndex = opts.memberIndex ?? false } #membersKey (): string { @@ -139,6 +163,15 @@ 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 [] @@ -192,7 +225,8 @@ 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 @@ -221,7 +255,7 @@ export class Registry { 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 @@ -240,10 +274,19 @@ export class Registry { } async addPodToDestination (destinationId: string, memberId: string): Promise { - await this.#redis.sadd(this.#destinationKey(destinationId), memberId) + if (this.#memberIndex) { + await this.#redis.multi().sadd(this.#destinationKey(destinationId), memberId).sadd(this.#ownedKey(memberId), destinationId).exec() + } else { + await this.#redis.sadd(this.#destinationKey(destinationId), memberId) + } if (this.#cache) this.#cache.delete(destinationId) } + // 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 { const count = await this.#redis.scard(this.#destinationKey(destinationId)) return count > 0 @@ -262,7 +305,15 @@ export class Registry { } async deregisterDestination (destinationId: string): Promise { - await this.#redis.del(this.#destinationKey(destinationId)) + const key = this.#destinationKey(destinationId) + if (this.#memberIndex) { + const members = await this.#redis.smembers(key) + const multi = this.#redis.multi().del(key) + for (const memberId of members) multi.srem(this.#ownedKey(memberId), destinationId) + await multi.exec() + } else { + await this.#redis.del(key) + } if (this.#cache) this.#cache.delete(destinationId) } diff --git a/test/member.test.ts b/test/member.test.ts index bc29aba..877a0a3 100644 --- a/test/member.test.ts +++ b/test/member.test.ts @@ -48,6 +48,26 @@ test('Member', async (t) => { 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 500f7d2..e58defb 100644 --- a/test/registry.test.ts +++ b/test/registry.test.ts @@ -248,6 +248,55 @@ test('Registry', async (t) => { } }) + 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 + await indexed.addPodToDestination('inst-owned-added', m2) + 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') From 66b8140fd351a7a748723fc73ac5b131f9966777 Mon Sep 17 00:00:00 2001 From: marcopiraccini Date: Sat, 14 Nov 2026 10:09:20 +0100 Subject: [PATCH 3/3] fixup Signed-off-by: marcopiraccini --- README.md | 4 +-- src/helpers/pick-and-register.ts | 15 +++++++-- src/member.ts | 16 ++++++++-- src/registry.ts | 55 +++++++++++++++++++++++--------- test/helpers.test.ts | 41 +++++++++++++++++++++++- test/member.test.ts | 8 +++++ test/registry.test.ts | 6 +++- 7 files changed, 121 insertions(+), 24 deletions(-) diff --git a/README.md b/README.md index b1d3499..badd3e0 100644 --- a/README.md +++ b/README.md @@ -176,7 +176,7 @@ if (lockRouting) { // Other primitives: 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 @@ -203,7 +203,7 @@ The index is exact only if every writer of the destinations maintains it, this l 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 script refuses it even when it was picked from a list read before it started draining: the coordinator then picks another pod. The destinations it already holds keep resolving to it until it deregisters or moves them. +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 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 275d952..d057799 100644 --- a/src/member.ts +++ b/src/member.ts @@ -25,6 +25,7 @@ export class Member { #ttl: number #getLoad: () => number #memberIndex: boolean + #draining: boolean constructor (opts: MemberOptions) { const { redis, owned } = connect(opts.redis, opts.redisOptions) @@ -36,6 +37,7 @@ export class Member { this.#ttl = opts.ttl ?? 30 this.#getLoad = opts.getLoad ?? (() => 0) this.#memberIndex = opts.memberIndex ?? false + this.#draining = false } get memberId (): string { @@ -79,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() @@ -95,9 +99,11 @@ export class Member { } // 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 flag survives the heartbeats, and goes with - // the record when it expires or the member deregisters. + // 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 { @@ -105,6 +111,10 @@ export class Member { } } + get draining (): boolean { + return this.#draining + } + async addToDestination (destinationId: string): Promise { if (this.#memberIndex) { await this.#redis.multi().sadd(this.#destinationKey(destinationId), this.#memberId).sadd(this.#ownedKey(), destinationId).exec() diff --git a/src/registry.ts b/src/registry.ts index 8dcd405..d6a30e0 100644 --- a/src/registry.ts +++ b/src/registry.ts @@ -104,6 +104,36 @@ const resolveScript = ` 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 @@ -116,6 +146,8 @@ type ResolveReply = interface ResolveCommand { 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 { @@ -130,6 +162,8 @@ export class Registry { 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' @@ -273,13 +307,12 @@ export class Registry { return { address: picked.address, memberId: picked.memberId, reassigned: false } } - async addPodToDestination (destinationId: string, memberId: string): Promise { - if (this.#memberIndex) { - await this.#redis.multi().sadd(this.#destinationKey(destinationId), memberId).sadd(this.#ownedKey(memberId), destinationId).exec() - } else { - 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 @@ -305,15 +338,7 @@ export class Registry { } async deregisterDestination (destinationId: string): Promise { - const key = this.#destinationKey(destinationId) - if (this.#memberIndex) { - const members = await this.#redis.smembers(key) - const multi = this.#redis.multi().del(key) - for (const memberId of members) multi.srem(this.#ownedKey(memberId), destinationId) - await multi.exec() - } else { - await this.#redis.del(key) - } + await this.#redis.coordinatorDeregister(this.#destinationKey(destinationId), this.#ownedPrefix(), destinationId) if (this.#cache) this.#cache.delete(destinationId) } 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 877a0a3..1748e83 100644 --- a/test/member.test.ts +++ b/test/member.test.ts @@ -43,7 +43,15 @@ test('Member', async (t) => { 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]]) }) diff --git a/test/registry.test.ts b/test/registry.test.ts index e58defb..400a73d 100644 --- a/test/registry.test.ts +++ b/test/registry.test.ts @@ -239,6 +239,10 @@ test('Registry', async (t) => { // 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') @@ -276,7 +280,7 @@ test('Registry', async (t) => { deepStrictEqual(await sharedRedis.sismember(ownedKey(m1), 'inst-owned-pruned'), 1) // Explicit bindings and deregistrations maintain it too - await indexed.addPodToDestination('inst-owned-added', m2) + 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)