Merge pull request #74 from mixer/v0.2.10

V0.2.10

changelog.md

@@ -1,3 +1,11 @@
+## 0.2.10 2017-05-05
+
+ - **feat**: update grpc with Node 10 support (see [#73](https://github.com/mixer/etcd3/pulls/73)) thanks to [@XadillaX](https://github.com/XadillaX)
+ - **feat**: add `lease.release()` to let leases expire automatically (see [#69](https://github.com/mixer/etcd3/issues/69))
+ - **bug**: update docs and throw if a lease TTL is not provided (see [#68](https://github.com/mixer/etcd3/issues/68))
+ - **bug**: forcefully terminate watch streams on close (see [#62](https://github.com/mixer/etcd3/issues/62))
+ - **bug**: reestablish watch streams if they're closed gracefully (see [#79](https://github.com/mixer/etcd3/issues/79))
+
 ## 0.2.9 2017-02-09
 
  - **bug**: lock to grpc@1.9.0 due to upstream regression (see [#59](https://github.com/mixer/etcd3/issues/59))

package.json

@@ -8,7 +8,7 @@
     "test": "npm-run-all --parallel test:lint test:unit",
     "test:unit": "mocha",
     "test:cover": "nyc mocha",
-    "test:lint": "tslint --type-check --project tsconfig.json '{src,test}/**/*.ts'",
+    "test:lint": "tslint --project tsconfig.json \"{src,test}/**/*.ts\"",
     "build:proto": "node ./bin/update-proto ./proto && node bin/generate-methods.js ./proto/rpc.proto > src/rpc.ts",
     "build:doc": "rm -rf docs && typedoc --gitRevision `git describe --abbrev=0 --tags` --exclude \"**/test/*\" --excludePrivate --out ./docs ./src/index.ts && node bin/tame-typedoc",
     "build:ts": "tsc && cp -R proto lib",
@@ -74,7 +74,7 @@
     "sinon": "^2.1.0",
     "ts-node": "^3.3.0",
     "tslint": "^5.7.0",
-    "tslint-microsoft-contrib": "^5.0.1",
+    "tslint-microsoft-contrib": "5.0.1",
     "typedoc": "^0.7.1",
     "typescript": "^2.7.1"
   },

src/connection-pool.ts

@@ -252,13 +252,12 @@ export class ConnectionPool implements ICallable {
       }));
     }
 
-    return Promise.all([
-      this.pool.pull(),
-      this.authenticator.getMetadata(),
-    ]).then(([host, metadata]) => {
-      const client = host.getServiceClient(service);
-      return { client, host, metadata };
-    });
+    return Promise.all([this.pool.pull(), this.authenticator.getMetadata()]).then(
+      ([host, metadata]) => {
+        const client = host.getServiceClient(service);
+        return { client, host, metadata };
+      },
+    );
   }
 
   /**

src/errors.ts

@@ -179,6 +179,18 @@ export class EtcdInvalidAuthTokenError extends Error {}
  */
 export class EtcdPermissionDeniedError extends Error {}
 
+/**
+ * EtcdWatchStreamEnded is emitted when a watch stream closes gracefully.
+ * This is an unexpected occurrence.
+ *
+ * @see https://github.com/mixer/etcd3/issues/72#issuecomment-386851271
+ */
+export class EtcdWatchStreamEnded extends Error {
+  constructor() {
+    super('The etcd watch stream was unexpectedly ended');
+  }
+}
+
 /**
  * An STMConflictError is thrown from the `SoftwareTransaction.transact`
  * if we continue to get conflicts and exceed the maximum number

src/index.ts

@@ -39,7 +39,7 @@ export class Etcd3 extends Namespace {
   public readonly cluster = new RPC.ClusterClient(this.pool);
 
   constructor(options: IOptions = { hosts: '127.0.0.1:2379' }) {
-    super(Buffer.from([]), new ConnectionPool(options));
+    super(Buffer.from([]), new ConnectionPool(options), options);
   }
 
   /**

src/lease.ts

@@ -70,7 +70,7 @@ const enum State {
  * const hostPrefix = 'available-hosts/';
  *
  * function grantLease() {
- *   const lease = client.lease();
+ *   const lease = client.lease(10); // set a TTL of 10 seconds
  *
  *   lease.on('lost', err => {
  *     console.log('We lost our lease as a result of this error:', err);
@@ -102,7 +102,7 @@ export class Lease extends EventEmitter {
   ) {
     super();
 
-    if (ttl < 1) {
+    if (!ttl || ttl < 1) {
       throw new Error(`The TTL in an etcd lease must be at least 1 second. Got: ${ttl}`);
     }
 
@@ -151,6 +151,15 @@ export class Lease extends EventEmitter {
     });
   }
 
+  /**
+   * releasePassively stops making heartbeats for the lease, and allows it
+   * to expire automatically when its TTL rolls around. Use `revoke()` to
+   * actively tell etcd to terminate the lease.
+   */
+  public release() {
+    this.close();
+  }
+
   /**
    * Put returns a put builder that operates within the current lease.
    */

src/namespace.ts

@@ -1,9 +1,10 @@
 import * as grpc from 'grpc';
 
 import * as Builder from './builder';
-import { ConnectionPool } from './connection-pool';
+import { ConnectionPool, defaultBackoffStrategy } from './connection-pool';
 import { Lease } from './lease';
 import { Lock } from './lock';
+import { IOptions } from './options';
 import { Rangable, Range } from './range';
 import * as RPC from './rpc';
 import { Isolation, ISTMOptions, SoftwareTransaction } from './stm';
@@ -35,9 +36,16 @@ export class Namespace {
   public readonly leaseClient = new RPC.LeaseClient(this.pool);
   public readonly watchClient = new RPC.WatchClient(this.pool);
   private readonly nsApplicator = new NSApplicator(this.prefix);
-  private readonly watchManager = new WatchManager(this.watchClient);
+  private readonly watchManager = new WatchManager(
+    this.watchClient,
+    this.options.backoffStrategy || defaultBackoffStrategy,
+  );
 
-  constructor(protected readonly prefix: Buffer, protected readonly pool: ConnectionPool) {}
+  protected constructor(
+    protected readonly prefix: Buffer,
+    protected readonly pool: ConnectionPool,
+    protected readonly options: IOptions,
+  ) {}
 
   /**
    * `.get()` starts a query to look up a single key from etcd.
@@ -140,6 +148,6 @@ export class Namespace {
    * with `user1/`. See the Namespace class for more details.
    */
   public namespace(prefix: string | Buffer): Namespace {
-    return new Namespace(Buffer.concat([this.prefix, toBuffer(prefix)]), this.pool);
+    return new Namespace(Buffer.concat([this.prefix, toBuffer(prefix)]), this.pool, this.options);
   }
 }

src/rpc.ts

@@ -26,6 +26,7 @@ export interface IResponseStream<T> {
 export interface IRequestStream<T> {
   write(item: T): void;
   end(): void;
+  cancel(): void;
 }
 
 export interface IDuplexStream<T, R> extends IRequestStream<T>, IResponseStream<R> {}

src/watch.ts

@@ -1,7 +1,13 @@
 import BigNumber from 'bignumber.js';
 import { EventEmitter } from 'events';
 
-import { castGrpcErrorMessage, ClientRuntimeError, EtcdError } from './errors';
+import { IBackoffStrategy } from './backoff/backoff';
+import {
+  castGrpcErrorMessage,
+  ClientRuntimeError,
+  EtcdError,
+  EtcdWatchStreamEnded,
+} from './errors';
 import { Rangable, Range } from './range';
 import * as RPC from './rpc';
 import { NSApplicator, onceEvent, toBuffer } from './util';
@@ -124,7 +130,7 @@ export class WatchManager {
    */
   private queue: null | AttachQueue;
 
-  constructor(private readonly client: RPC.WatchClient) {}
+  constructor(private readonly client: RPC.WatchClient, private backoff: IBackoffStrategy) {}
 
   /**
    * Attach registers the watcher on the connection.
@@ -215,7 +221,8 @@ export class WatchManager {
         this.queue = new AttachQueue(stream);
         this.stream = stream
           .on('data', res => this.handleResponse(res))
-          .on('error', err => this.handleError(err));
+          .on('error', err => this.handleError(err))
+          .on('end', () => this.handleError(new EtcdWatchStreamEnded()));
 
         // possible watchers are remove while we're connecting.
         if (this.watchers.length === 0) {
@@ -238,7 +245,7 @@ export class WatchManager {
       throw new ClientRuntimeError('Cannot call destroyStream() with active watchers');
     }
 
-    this.getStream().end();
+    this.getStream().cancel();
     this.queue!.destroy();
   }
 
@@ -249,7 +256,7 @@ export class WatchManager {
   private handleError(err: Error) {
     if (this.state === State.Connected) {
       this.queue!.destroy();
-      this.getStream().end();
+      this.getStream().cancel();
     }
     this.state = State.Idle;
 
@@ -258,7 +265,13 @@ export class WatchManager {
       (<{ id: null }>watcher).id = null;
     });
 
-    this.establishStream();
+    setTimeout(() => {
+      if (this.state === State.Idle) {
+        this.establishStream();
+      }
+    }, this.backoff.getDelay());
+
+    this.backoff = this.backoff.next();
   }
 
   /**
@@ -288,6 +301,8 @@ export class WatchManager {
    * Dispatches some watch response on the event stream.
    */
   private handleResponse(res: RPC.IWatchResponse) {
+    this.backoff = this.backoff.reset();
+
     if (res.created) {
       this.queue!.handleCreate(res);
       return;

test/lease.test.ts

@@ -28,8 +28,9 @@ describe('lease()', () => {
     }
   });
 
-  it('throws if trying to use too short of a ttl', () => {
+  it('throws if trying to use too short of a ttl, or an undefined ttl', () => {
     expect(() => client.lease(0)).to.throw(/must be at least 1 second/);
+    expect(() => (<any>client.lease)()).to.throw(/must be at least 1 second/);
   });
 
   it('reports a loss and errors if the client is invalid', async () => {
@@ -151,6 +152,13 @@ describe('lease()', () => {
       expect(res.TTL).to.equal('60');
     });
 
+    it('stops touching the lease if released passively', async () => {
+      const kaFired = watchEmission('keepaliveFired');
+      lease.release();
+      clock.tick(20000);
+      expect(kaFired.fired).to.be.false;
+    });
+
     it('tears down if the lease gets revoked', async () => {
       await client.leaseClient.leaseRevoke({ ID: await lease.grant() });
       clock.tick(20000);

test/watch.test.ts

@@ -5,7 +5,7 @@ import { Etcd3, IKeyValue, IWatchResponse, Watcher } from '../src';
 import { onceEvent } from '../src/util';
 import { createTestClientAndKeys, getOptions, proxy, tearDownTestClient } from './util';
 
-describe('watch', () => {
+describe('watch()', () => {
   let client: Etcd3;
 
   beforeEach(async () => {