feat!: remove legacy ack deadline parameters, and move maxExtension into subscriber options

Author: feywind

src/default-options.ts

@@ -34,16 +34,16 @@ export const defaultOptions = {
     // The maximum length of time a message's lease will be extended by.
     maxAckDeadline: Duration.from({minutes: 10}),
 
-    // The maximum number of minutes that a message's lease will ever
+    // The maximum amount of time that a message's lease will ever
     // be extended.
-    maxExtensionMinutes: 60,
+    maxExtensionTime: Duration.from({minutes: 60}),
 
     // The maximum number of subscription streams/threads that will ever
     // be opened.
     maxStreams: 5,
 
     // The starting number of seconds that ack deadlines will be extended.
-    ackDeadline: 10,
+    startingAckDeadline: Duration.from({seconds: 10}),
   },
 
   publish: {

src/lease-manager.ts

@@ -23,10 +23,6 @@ export interface FlowControlOptions {
   allowExcessMessages?: boolean;
   maxBytes?: number;
   maxMessages?: number;
-  maxExtensionMinutes?: number;
-
-  /** @deprecated Use maxExtensionMinutes. */
-  maxExtension?: number;
 }
 
 /**
@@ -194,28 +190,9 @@ export class LeaseManager extends EventEmitter {
    * @private
    */
   setOptions(options: FlowControlOptions): void {
-    // Convert the old deprecated maxExtension to avoid breaking clients,
-    // but allow only one.
-    if (
-      options.maxExtension !== undefined &&
-      options.maxExtensionMinutes !== undefined
-    ) {
-      throw new RangeError(
-        'Only one of "maxExtension" or "maxExtensionMinutes" may be set for subscriber lease management options',
-      );
-    }
-    if (
-      options.maxExtension !== undefined &&
-      options.maxExtensionMinutes === undefined
-    ) {
-      options.maxExtensionMinutes = options.maxExtension / 60;
-      delete options.maxExtension;
-    }
-
     const defaults: FlowControlOptions = {
       allowExcessMessages: true,
       maxBytes: defaultOptions.subscription.maxOutstandingBytes,
-      maxExtensionMinutes: defaultOptions.subscription.maxExtensionMinutes,
       maxMessages: defaultOptions.subscription.maxOutstandingMessages,
     };
 
@@ -260,19 +237,20 @@ export class LeaseManager extends EventEmitter {
    * @private
    */
   private _extendDeadlines(): void {
-    const deadline = this._subscriber.ackDeadline;
+    const deadline = Duration.from({seconds: this._subscriber.ackDeadline});
+    const maxExtensionMinutes =
+      this._subscriber.maxExtensionTime.totalOf('minute');
 
     for (const message of this._messages) {
       // Lifespan here is in minutes.
       const lifespan = (Date.now() - message.received) / (60 * 1000);
 
-      if (lifespan < this._options.maxExtensionMinutes!) {
-        const deadlineDuration = Duration.from({seconds: deadline});
-        message.subSpans.modAckStart(deadlineDuration, false);
+      if (lifespan < maxExtensionMinutes) {
+        message.subSpans.modAckStart(deadline, false);
 
         if (this._subscriber.isExactlyOnceDelivery) {
           message
-            .modAckWithResponse(deadline)
+            .modAckWithResponse(deadline.totalOf('second'))
             .catch(e => {
               // In the case of a permanent failure (temporary failures are retried),
               // we need to stop trying to lease-manage the message.
@@ -283,8 +261,8 @@ export class LeaseManager extends EventEmitter {
               message.subSpans.modAckEnd();
             });
         } else {
-          message.modAck(deadline);
-          message.subSpans.modAckStart(deadlineDuration, false);
+          message.modAck(deadline.totalOf('second'));
+          message.subSpans.modAckStart(deadline, false);
         }
       } else {
         this.remove(message);

src/subscriber.ts

@@ -560,6 +560,8 @@ export class Message implements tracing.MessageWithAttributes {
  *     ever have, while it's under library control.
  * @property {Duration} [maxAckDeadline] The maximum time that ackDeadline should
  *     ever have, while it's under library control.
+ * @property {Duration} [maxExtensionTime] The maximum time that ackDeadline should
+ *     ever have, while it's under library control.
  * @property {BatchOptions} [batching] Request batching options; this is for
  *     batching acks and modacks being sent back to the server.
  * @property {FlowControlOptions} [flowControl] Flow control options.
@@ -569,11 +571,9 @@ export class Message implements tracing.MessageWithAttributes {
  * @property {MessageStreamOptions} [streamingOptions] Streaming options.
  */
 export interface SubscriberOptions {
-  /** @deprecated Use minAckDeadline and maxAckDeadline. */
-  ackDeadline?: number;
-
   minAckDeadline?: Duration;
   maxAckDeadline?: Duration;
+  maxExtensionTime?: Duration;
   batching?: BatchOptions;
   flowControl?: FlowControlOptions;
   useLegacyFlowControl?: boolean;
@@ -597,6 +597,7 @@ export class Subscriber extends EventEmitter {
   maxBytes: number;
   useLegacyFlowControl: boolean;
   isOpen: boolean;
+  maxExtensionTime: Duration;
   private _acks!: AckQueue;
   private _histogram: Histogram;
   private _inventory!: LeaseManager;
@@ -612,9 +613,11 @@ export class Subscriber extends EventEmitter {
   constructor(subscription: Subscription, options = {}) {
     super();
 
-    this.ackDeadline = defaultOptions.subscription.ackDeadline;
+    this.ackDeadline =
+      defaultOptions.subscription.startingAckDeadline.totalOf('second');
     this.maxMessages = defaultOptions.subscription.maxOutstandingMessages;
     this.maxBytes = defaultOptions.subscription.maxOutstandingBytes;
+    this.maxExtensionTime = defaultOptions.subscription.maxExtensionTime;
     this.useLegacyFlowControl = false;
     this.isOpen = false;
     this._histogram = new Histogram({min: 10, max: 600});
@@ -960,15 +963,6 @@ export class Subscriber extends EventEmitter {
   setOptions(options: SubscriberOptions): void {
     this._options = options;
 
-    // The user-set ackDeadline value basically pegs the extension time.
-    // We'll emulate it by overwriting min/max.
-    const passedAckDeadline = options.ackDeadline;
-    if (passedAckDeadline !== undefined) {
-      this.ackDeadline = passedAckDeadline;
-      options.minAckDeadline = Duration.from({seconds: passedAckDeadline});
-      options.maxAckDeadline = Duration.from({seconds: passedAckDeadline});
-    }
-
     this.useLegacyFlowControl = options.useLegacyFlowControl || false;
     if (options.flowControl) {
       this.maxMessages =
@@ -998,6 +992,18 @@ export class Subscriber extends EventEmitter {
     if (this._inventory) {
       this._inventory.setOptions(this._options.flowControl!);
     }
+
+    this.updateAckDeadline();
+  }
+
+  /**
+   * Retrieves our effective options. This is mostly for unit test use.
+   *
+   * @private
+   * @returns {SubscriberOptions} The options.
+   */
+  getOptions(): SubscriberOptions {
+    return this._options;
   }
 
   /**

system-test/pubsub.ts

@@ -31,6 +31,7 @@ import {
   SchemaTypes,
   SchemaViews,
   ISchema,
+  Duration,
 } from '../src';
 import {Policy, IamPermissionsMap} from '../src/iam';
 import {MessageOptions} from '../src/topic';
@@ -428,10 +429,12 @@ describe('pubsub', () => {
     const SUB_NAMES = [generateSubName(), generateSubName()];
     const SUB_DETACH_NAME = generateSubForDetach();
 
+    const thirty = Duration.from({minutes: 30});
+    const sixty = Duration.from({minutes: 60});
     const SUBSCRIPTIONS = [
-      topic.subscription(SUB_NAMES[0], {ackDeadline: 30}),
-      topic.subscription(SUB_NAMES[1], {ackDeadline: 60}),
-      topic.subscription(SUB_DETACH_NAME, {ackDeadline: 30}),
+      topic.subscription(SUB_NAMES[0], {minAckDeadline: thirty, maxAckDeadline: thirty}),
+      topic.subscription(SUB_NAMES[1], {minAckDeadline: sixty, maxAckDeadline: sixty}),
+      topic.subscription(SUB_DETACH_NAME, {minAckDeadline: thirty, maxAckDeadline: thirty}),
     ];
 
     before(async () => {

test/lease-manager.ts

@@ -31,6 +31,7 @@ import {
 } from '../src/subscriber';
 import {defaultOptions} from '../src/default-options';
 import {TestUtils} from './test-utils';
+import {Duration} from '../src';
 
 const FREE_MEM = 9376387072;
 const fakeos = {
@@ -41,6 +42,7 @@ class FakeSubscriber extends EventEmitter {
   ackDeadline = 10;
   isOpen = true;
   modAckLatency = 2000;
+  maxExtensionTime = Duration.from({minutes: 60});
   async modAck(): Promise<void> {}
   async modAckWithResponse(): Promise<AckResponse> {
     return AckResponses.Success;
@@ -286,30 +288,12 @@ describe('LeaseManager', () => {
         });
       });
 
-      it('should properly convert any legacy maxExtension values', () => {
-        const maxExtension = 60 * 1000;
-        leaseManager.setOptions({maxExtension});
-        // eslint-disable-next-line @typescript-eslint/no-explicit-any
-        const options = (leaseManager as any)._options;
-        assert.strictEqual(options.maxExtensionMinutes, maxExtension / 60);
-        assert.strictEqual(options.maxExtension, undefined);
-      });
-
-      it('should not allow both maxExtension and maxExtensionMinutes', () => {
-        assert.throws(() => {
-          leaseManager.setOptions({
-            maxExtension: 10,
-            maxExtensionMinutes: 10,
-          });
-        });
-      });
-
       it('should remove any messages that pass the maxExtensionMinutes value', () => {
         const maxExtensionSeconds = (expectedTimeout - 100) / 1000;
         const badMessages = [new FakeMessage(), new FakeMessage()];
 
-        leaseManager.setOptions({
-          maxExtensionMinutes: maxExtensionSeconds / 60,
+        subscriber.maxExtensionTime = Duration.from({
+          seconds: maxExtensionSeconds,
         });
         badMessages.forEach(message =>
           leaseManager.add(message as {} as Message),
@@ -339,9 +323,7 @@ describe('LeaseManager', () => {
       it('should remove and ackFailed any messages that fail to ack', done => {
         (subscriber as unknown as FakeSubscriber).isExactlyOnceDelivery = true;
 
-        leaseManager.setOptions({
-          maxExtensionMinutes: 600,
-        });
+        subscriber.maxExtensionTime = Duration.from({minutes: 600});
 
         const goodMessage = new FakeMessage();
 

test/subscriber.ts

@@ -259,12 +259,14 @@ describe('Subscriber', () => {
     });
 
     it('should set any options passed in', () => {
-      const stub = sandbox.stub(Subscriber.prototype, 'setOptions');
-      const fakeOptions = {};
-      new Subscriber(subscription, fakeOptions);
-
-      const [options] = stub.lastCall.args;
-      assert.strictEqual(options, fakeOptions);
+      const options = {
+        streamingOptions: {},
+      };
+      const subscriber = new Subscriber(subscription, options);
+      assert.strictEqual(
+        subscriber.getOptions().streamingOptions,
+        options.streamingOptions,
+      );
     });
   });
 
@@ -464,7 +466,7 @@ describe('Subscriber', () => {
     });
 
     it('should default to 60s min for exactly-once delivery subscriptions', async () => {
-      subscriber.subscriptionProperties = {exactlyOnceDeliveryEnabled: true};
+      subscriber.setSubscriptionProperties({exactlyOnceDeliveryEnabled: true});
 
       const histogram: FakeHistogram = stubs.get('histogram');
       const now = Date.now();
@@ -501,8 +503,10 @@ describe('Subscriber', () => {
       sandbox.stub(histogram, 'add').throws();
       sandbox.stub(histogram, 'percentile').throws();
 
+      const deadlineTime = Duration.from({seconds: ackDeadline});
       subscriber.setOptions({
-        ackDeadline,
+        minAckDeadline: deadlineTime,
+        maxAckDeadline: deadlineTime,
         flowControl: {maxMessages: maxMessages, maxBytes: maxBytes},
       });
       void subscriber.ack(message);
@@ -847,10 +851,13 @@ describe('Subscriber', () => {
     beforeEach(() => subscriber.close());
 
     it('should capture the ackDeadline', () => {
-      const ackDeadline = 1232;
+      const ackDeadline = Duration.from({seconds: 1232});
 
-      subscriber.setOptions({ackDeadline});
-      assert.strictEqual(subscriber.ackDeadline, ackDeadline);
+      subscriber.setOptions({
+        minAckDeadline: ackDeadline,
+        maxAckDeadline: ackDeadline,
+      });
+      assert.strictEqual(subscriber.ackDeadline, ackDeadline.totalOf('second'));
     });
 
     it('should not set maxStreams higher than maxMessages', () => {