fix: elections sometimes electing >1 leader

Fixes #176
Fixes #158

Author: connor4312

changelog.md

@@ -1,5 +1,10 @@
 # Changelog
 
+# 1.2.1 2023-07-30
+
+- **fix:** elections sometimes electing >1 leader (see [#176](https://github.com/microsoft/etcd3/issues/176))
+- **fix:** a race condition in Host.resetAllServices (see [#182](https://github.com/microsoft/etcd3/issues/182))
+
 ## 1.2.0 2023-07-28
 
 - **fix:** leases revoked or released before grant completes leaking

src/election.ts

@@ -8,7 +8,7 @@ import { ClientRuntimeError, NotCampaigningError } from './errors';
 import { Lease } from './lease';
 import { Namespace } from './namespace';
 import { IKeyValue } from './rpc';
-import { getDeferred, IDeferred, toBuffer } from './util';
+import { IDeferred, getDeferred, toBuffer } from './util';
 
 const UnsetCurrent = Symbol('unset');
 
@@ -331,20 +331,28 @@ export class Campaign extends EventEmitter {
   }
 
   private async waitForElected(revision: string) {
-    // find last created before this one
-    const lastRevision = new BigNumber(revision).minus(1).toString();
-    const result = await this.namespace
-      .getAll()
-      .maxCreateRevision(lastRevision)
-      .sort('Create', 'Descend')
-      .limit(1)
-      .exec();
-
-    // wait for all older keys to be deleted for us to become the leader
-    await waitForDeletes(
-      this.namespace,
-      result.kvs.map(k => k.key),
-    );
+    while (this.keyRevision !== ResignedCampaign) {
+      // find last created before this one
+      const lastRevision = new BigNumber(revision).minus(1).toString();
+      const result = await this.namespace
+        .getAll()
+        .maxCreateRevision(lastRevision)
+        .sort('Create', 'Descend')
+        .limit(1)
+        .exec();
+
+      if (result.kvs.length === 0) {
+        return;
+      }
+
+      this.emit('_isWaiting'); // internal event used to sync unit tests
+
+      // wait for all it to be deleted for us to become the leader
+      await waitForDeletes(
+        this.namespace,
+        result.kvs.map(k => k.key),
+      );
+    }
   }
 }
 

src/test/election.test.ts

@@ -4,7 +4,7 @@ import { take } from 'rxjs/operators';
 import { Election, Etcd3 } from '../';
 import { Campaign } from '../election';
 import { NotCampaigningError } from '../errors';
-import { delay, getDeferred } from '../util';
+import { delay, getDeferred, onceEvent } from '../util';
 import { getOptions, tearDownTestClient } from './util';
 
 const sleep = (t: number) => new Promise(resolve => setTimeout(resolve, t));
@@ -192,4 +192,38 @@ describe('election', () => {
       await observer.cancel();
     });
   });
+
+  it('fixes #176', async function () {
+    const observer1 = await election.observe();
+
+    const client2 = new Etcd3(getOptions());
+    const election2 = client2.election('test-election', 1);
+    const observer2 = await election2.observe();
+    const campaign2 = election2.campaign('candidate2');
+    await onceEvent(campaign2, '_isWaiting');
+
+    const client3 = new Etcd3(getOptions());
+    const election3 = client3.election('test-election', 1);
+    const observer3 = await election3.observe();
+    const campaign3 = election3.campaign('candidate3');
+    await onceEvent(campaign3, '_isWaiting');
+
+    expect(observer1.leader()).to.equal('candidate');
+    expect(observer2.leader()).to.equal('candidate');
+    expect(observer3.leader()).to.equal('candidate');
+
+    const changes: string[] = [];
+    campaign.on('elected', () => changes.push('leader is now 1'));
+    campaign3.on('elected', () => changes.push('leader is now 3'));
+
+    await campaign2.resign();
+    await delay(1000); // give others a chance to see the change, if any
+
+    expect(observer1.leader()).to.equal('candidate');
+    expect(observer3.leader()).to.equal('candidate');
+    expect(changes).to.be.empty;
+
+    client2.close();
+    client3.close();
+  });
 });