Sperren: Timeout eines Wartenden hebt die Sperre nicht auf
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
679f996597
commit
ea71b3f1c3
2 files changed
+52
-15
No files matched your search
@@ -2,35 +2,60 @@
|
||||
* Sperren je Schluessel (z. B. Kundennummer, Ticket-Id), damit parallele Anfragen fuer
|
||||
* dieselbe Ressource nacheinander laufen, waehrend Anfragen zu anderen Schluesseln parallel
|
||||
* bleiben. Ein Aufruf kann mehrere Schluessel auf einmal sperren (alle muessen frei werden).
|
||||
* waitMs begrenzt, wie lange auf die aktuellen Inhaber gewartet wird (Default: unbegrenzt);
|
||||
* nach Ablauf laeuft fn trotzdem los, ohne weiter zu warten.
|
||||
* waitMs begrenzt nur, wie lange DIESER Aufruf auf die aktuellen Inhaber wartet, bevor er
|
||||
* trotzdem losläuft (Default: unbegrenzt) - spaetere Aufrufe warten trotzdem auf die gesamte
|
||||
* Kette bis hierhin, einschliesslich eines Inhabers, den dieser Aufruf per waitMs uebersprungen
|
||||
* hat.
|
||||
*
|
||||
* Umsetzung: je Schluessel eine Promise-Kette (das "tail") in einer modulweiten Map.
|
||||
* Umsetzung: je Schluessel eine Promise-Kette (das "tail") in einer modulweiten Map. Der
|
||||
* gespeicherte Wert ist immer "die gesamte Kette bis einschliesslich dieses Aufrufs ist fertig"
|
||||
* (Promise.allSettled der vorherigen tails + der eigene Abschluss) - nie nur der eigene
|
||||
* Abschluss allein. Sonst wuerde ein Aufruf, der per waitMs vorzeitig losgelaufen ist, beim
|
||||
* Aufraeumen die Spur eines noch laufenden, unbegrenzt wartenden frueheren Inhabers loeschen,
|
||||
* und ein dritter Aufruf wuerde faelschlich gar nicht mehr warten.
|
||||
*/
|
||||
const tailsByKey = new Map()
|
||||
|
||||
// setTimeout(fn, Infinity) wuerde Node zu 1 ms clampen (samt Warnung) - hier bewusst
|
||||
// eine Promise, die nie aufloest, damit "unbegrenzt warten" auch wirklich unbegrenzt ist.
|
||||
const sleep = (ms) =>
|
||||
ms === Infinity ? new Promise(() => {}) : new Promise((resolve) => setTimeout(resolve, ms))
|
||||
// setTimeout(fn, Infinity) wuerde Node zu 1 ms clampen (samt Warnung) - hier bewusst eine
|
||||
// Promise ohne echten Timer, damit "unbegrenzt warten" auch wirklich unbegrenzt ist und nichts
|
||||
// aufgeraeumt werden muss. Bei endlichem waitMs wird der Timer ueber cancel() wieder entfernt,
|
||||
// sobald das Rennen entschieden ist - sonst haelt ein liegen gebliebener Timer z. B.
|
||||
// `node --test` unnoetig am Leben.
|
||||
function waitMsTimeout(waitMs) {
|
||||
if (waitMs === Infinity) return { promise: new Promise(() => {}), cancel: () => {} }
|
||||
let timer
|
||||
const promise = new Promise((resolve) => {
|
||||
timer = setTimeout(resolve, waitMs)
|
||||
})
|
||||
return { promise, cancel: () => clearTimeout(timer) }
|
||||
}
|
||||
|
||||
export async function withLocks(keys, fn, { waitMs = Infinity } = {}) {
|
||||
const previousTails = keys.map((key) => tailsByKey.get(key)).filter(Boolean)
|
||||
|
||||
let releaseTail
|
||||
const tail = new Promise((resolve) => {
|
||||
releaseTail = resolve
|
||||
let resolveOwnDone
|
||||
const ownDone = new Promise((resolve) => {
|
||||
resolveOwnDone = resolve
|
||||
})
|
||||
// Gespeichert wird die gesamte Kette (siehe Kommentar oben), nicht nur ownDone.
|
||||
const tail = Promise.allSettled([...previousTails, ownDone])
|
||||
for (const key of keys) tailsByKey.set(key, tail)
|
||||
tail.then(() => {
|
||||
for (const key of keys) {
|
||||
if (tailsByKey.get(key) === tail) tailsByKey.delete(key)
|
||||
}
|
||||
})
|
||||
|
||||
await Promise.race([Promise.allSettled(previousTails), sleep(waitMs)])
|
||||
const { promise: timeout, cancel: cancelTimeout } = waitMsTimeout(waitMs)
|
||||
try {
|
||||
await Promise.race([Promise.allSettled(previousTails), timeout])
|
||||
} finally {
|
||||
cancelTimeout()
|
||||
}
|
||||
|
||||
try {
|
||||
return await fn()
|
||||
} finally {
|
||||
releaseTail()
|
||||
for (const key of keys) {
|
||||
if (tailsByKey.get(key) === tail) tailsByKey.delete(key)
|
||||
}
|
||||
resolveOwnDone()
|
||||
}
|
||||
}
|
||||
@@ -37,3 +37,15 @@ test('Fehler geben die Sperre frei', async () => {
|
||||
await assert.rejects(withLocks(['a'], async () => { throw new Error('x') }))
|
||||
assert.equal(await withLocks(['a'], async () => 42), 42)
|
||||
})
|
||||
test('Timeout eines Wartenden hebt die Sperre nicht auf', async () => {
|
||||
const call3 = (async () => {
|
||||
await sleep(100)
|
||||
await withLocks(['a'], job('3', 10))
|
||||
})()
|
||||
await Promise.all([
|
||||
withLocks(['a'], job('1', 300)),
|
||||
withLocks(['a'], job('2', 10), { waitMs: 50 }),
|
||||
call3,
|
||||
])
|
||||
assert.ok(log.indexOf('3:start') > log.indexOf('1:end'), `log: ${log.join(',')}`)
|
||||
})
|
||||
Reference in new issue
Block a user