diff --git a/src/backends/file/leaser.ts b/src/backends/file/leaser.ts index 262ff64..4b86a26 100644 --- a/src/backends/file/leaser.ts +++ b/src/backends/file/leaser.ts @@ -8,7 +8,7 @@ */ import { createHash } from 'node:crypto'; -import { open, readFile, rename, rm, writeFile } from 'node:fs/promises'; +import { link, readFile, rename, rm, writeFile } from 'node:fs/promises'; import nodePath from 'node:path'; import { @@ -21,6 +21,9 @@ import { } from '../../leaser.ts'; import { FileReplicaError } from '../../errors.ts'; +// Disambiguates temp files written in the same millisecond by this process. +let tmpSeq = 0; + function sha256(data: Uint8Array): string { return createHash('sha256').update(data).digest('hex'); } @@ -82,10 +85,13 @@ export class FileLeaser implements Leaser { const bytes = encodeLease(lease); if (existing === null) { - // Exclusive create so we do not clobber a concurrent acquirer. - let fh; + // Exclusive create so we do not clobber a concurrent acquirer. Write a + // temp file and hard-link it into place so the lock appears atomically + // with its full contents; a racing reader never sees an empty file. + const tmp = `${this.lockPath}.tmp-${process.pid}-${Date.now()}-${tmpSeq++}`; + await writeFile(tmp, bytes); try { - fh = await open(this.lockPath, 'wx'); + await link(tmp, this.lockPath); } catch (err) { const code = (err as NodeJS.ErrnoException).code; if (code === 'EEXIST') { @@ -97,12 +103,12 @@ export class FileLeaser implements Leaser { return this.acquireLease(); } throw err; + } finally { + await rm(tmp, { force: true }); } - await fh.write(bytes); - await fh.close(); } else { // Existing lease expired: atomically replace it. - const tmp = `${this.lockPath}.tmp-${process.pid}-${Date.now()}`; + const tmp = `${this.lockPath}.tmp-${process.pid}-${Date.now()}-${tmpSeq++}`; await writeFile(tmp, bytes); await rename(tmp, this.lockPath); } @@ -118,7 +124,7 @@ export class FileLeaser implements Leaser { const renewed = new Lease(lease.generation, Date.now() + this.ttlMs, lease.owner); const bytes = encodeLease(renewed); - const tmp = `${this.lockPath}.tmp-${process.pid}-${Date.now()}`; + const tmp = `${this.lockPath}.tmp-${process.pid}-${Date.now()}-${tmpSeq++}`; await writeFile(tmp, bytes); await rename(tmp, this.lockPath); diff --git a/src/replica.ts b/src/replica.ts index 73369c0..f20dae0 100644 --- a/src/replica.ts +++ b/src/replica.ts @@ -306,6 +306,9 @@ export class Replica { set leaser(leaser: Leaser | null) { this.leaserValue = leaser; + // Mirror the constructor: a newly set leaser is not held until acquired. + // While running, {@link run} owns the primary state. + if (!this.runActive) this.leasePrimary = leaser === null; } /** diff --git a/test/integration/leasing.test.ts b/test/integration/leasing.test.ts index 455f5f3..d7a93b5 100644 --- a/test/integration/leasing.test.ts +++ b/test/integration/leasing.test.ts @@ -83,6 +83,30 @@ describe('FileLeaser', () => { void expired; await rm(dir, { recursive: true, force: true }); }); + + test('concurrent acquires: exactly one wins, the rest see LeaseExistsError', async () => { + for (let i = 0; i < 50; i++) { + const dir = makeTempDir('lease-race-'); + const results = await Promise.allSettled( + Array.from({ length: 4 }, () => new FileLeaser(dir).acquireLease()), + ); + const won = results.filter((r) => r.status === 'fulfilled'); + assert.equal(won.length, 1); + for (const r of results) { + if (r.status === 'rejected') assert.ok(r.reason instanceof LeaseExistsError, String(r.reason)); + } + await rm(dir, { recursive: true, force: true }); + } + }); +}); + +test('setting a leaser makes the replica a follower until it acquires', () => { + const replica = new Replica('/nonexistent.db', new FileReplicaClient(makeTempDir('lease-set-'))); + assert.ok(replica.isPrimary()); + replica.leaser = new FileLeaser(makeTempDir('lease-set-')); + assert.ok(!replica.isPrimary()); + replica.leaser = null; + assert.ok(replica.isPrimary()); }); describe('Replica.run() with a leaser (primary/follower)', () => {