Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fix: Added connection error timeout #485

Closed
wants to merge 5 commits into from
Closed

fix: Added connection error timeout #485

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
3110742
fix: Added connection error timeout
fraserbenjamin Apr 4, 2024
275fbce
fix: Reverted package.json
fraserbenjamin Apr 4, 2024
9ecf875
Removed additional error type
fraserbenjamin Apr 6, 2024
b9554a7
Removed unused type
fraserbenjamin Apr 6, 2024
b1fc753
Added QueueDoesNotExist catch.
fraserbenjamin Apr 12, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 7 additions & 7 deletions src/consumer.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,7 @@ export class Consumer extends TypedEventEmitter {
private visibilityTimeout: number;
private terminateVisibilityTimeout: boolean;
private waitTimeSeconds: number;
private authenticationErrorTimeout: number;
private connectionErrorTimeout: number;
private pollingWaitTimeMs: number;
private pollingCompleteWaitTimeMs: number;
private heartbeatInterval: number;
Expand All @@ -76,8 +76,8 @@ export class Consumer extends TypedEventEmitter {
options.terminateVisibilityTimeout || false;
this.heartbeatInterval = options.heartbeatInterval;
this.waitTimeSeconds = options.waitTimeSeconds ?? 20;
this.authenticationErrorTimeout =
options.authenticationErrorTimeout ?? 10000;
this.connectionErrorTimeout =
options.connectionErrorTimeout ?? 10000;
this.pollingWaitTimeMs = options.pollingWaitTimeMs ?? 0;
this.pollingCompleteWaitTimeMs = options.pollingCompleteWaitTimeMs ?? 0;
this.shouldDeleteMessages = options.shouldDeleteMessages ?? true;
Expand Down Expand Up @@ -249,14 +249,14 @@ export class Consumer extends TypedEventEmitter {
)
.catch((err): void => {
this.emitError(err);

if (isConnectionError(err)) {
logger.debug("authentication_error", {
logger.debug("connection_error", {
detail:
"There was an authentication error. Pausing before retrying.",
`${err.code}: There was an connection error. Pausing before retrying.`,
});
currentPollingTimeout = this.authenticationErrorTimeout;
currentPollingTimeout = this.connectionErrorTimeout;
}
return;
})
.then((): void => {
if (this.pollingTimeoutId) {
Expand Down
5 changes: 4 additions & 1 deletion src/errors.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,10 +44,13 @@ function isConnectionError(err: Error): boolean {
if (err instanceof SQSError) {
return (
err.statusCode === 403 ||
err.name === "SQSError" ||
fraserbenjamin marked this conversation as resolved.
Show resolved Hide resolved
err.code === "CredentialsError" ||
err.code === "UnknownEndpoint" ||
err.code === "AWS.SimpleQueueService.NonExistentQueue" ||
fraserbenjamin marked this conversation as resolved.
Show resolved Hide resolved
err.code === "CredentialsProviderError"
err.code === "CredentialsProviderError" ||
err.code === "QueueDoesNotExist" ||
err.code === "InvalidAddress"
);
}
return false;
Expand Down
4 changes: 2 additions & 2 deletions src/types.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,10 +40,10 @@ export interface ConsumerOptions {
*/
waitTimeSeconds?: number;
/**
* The duration (in milliseconds) to wait before retrying after an authentication error.
* The duration (in milliseconds) to wait before retrying after a connection or authentication error.
* @defaultvalue `10000`
*/
authenticationErrorTimeout?: number;
connectionErrorTimeout?: number;
/**
* The duration (in milliseconds) to wait before repolling the queue.
* @defaultvalue `0`
Expand Down
63 changes: 42 additions & 21 deletions test/tests/consumer.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ import { logger } from "../../src/logger.js";

const sandbox = sinon.createSandbox();

const AUTHENTICATION_ERROR_TIMEOUT = 20;
const CONNECTION_ERROR_TIMEOUT = 20;
const POLLING_TIMEOUT = 100;
const QUEUE_URL = "some-queue-url";
const REGION = "some-region";
Expand Down Expand Up @@ -86,7 +86,7 @@ describe("Consumer", () => {
region: REGION,
handleMessage,
sqs,
authenticationErrorTimeout: AUTHENTICATION_ERROR_TIMEOUT,
connectionErrorTimeout: CONNECTION_ERROR_TIMEOUT,
});
});

Expand Down Expand Up @@ -256,7 +256,7 @@ describe("Consumer", () => {
new Promise((resolve) => setTimeout(resolve, 1000)),
handleMessageTimeout,
sqs,
authenticationErrorTimeout: AUTHENTICATION_ERROR_TIMEOUT,
connectionErrorTimeout: CONNECTION_ERROR_TIMEOUT,
});

consumer.start();
Expand All @@ -281,7 +281,7 @@ describe("Consumer", () => {
throw new Error("unexpected parsing error");
},
sqs,
authenticationErrorTimeout: AUTHENTICATION_ERROR_TIMEOUT,
connectionErrorTimeout: CONNECTION_ERROR_TIMEOUT,
});

consumer.start();
Expand Down Expand Up @@ -315,7 +315,7 @@ describe("Consumer", () => {
throw new CustomError("unexpected parsing error");
},
sqs,
authenticationErrorTimeout: AUTHENTICATION_ERROR_TIMEOUT,
connectionErrorTimeout: CONNECTION_ERROR_TIMEOUT,
});

consumer.start();
Expand All @@ -339,7 +339,7 @@ describe("Consumer", () => {
throw customError;
},
sqs,
authenticationErrorTimeout: AUTHENTICATION_ERROR_TIMEOUT,
connectionErrorTimeout: CONNECTION_ERROR_TIMEOUT,
});

consumer.start();
Expand Down Expand Up @@ -418,7 +418,7 @@ describe("Consumer", () => {
consumer.on("error", errorListener);

consumer.start();
await clock.tickAsync(AUTHENTICATION_ERROR_TIMEOUT);
await clock.tickAsync(CONNECTION_ERROR_TIMEOUT);
consumer.stop();

sandbox.assert.calledTwice(errorListener);
Expand All @@ -438,7 +438,7 @@ describe("Consumer", () => {
consumer.on("error", errorListener);

consumer.start();
await clock.tickAsync(AUTHENTICATION_ERROR_TIMEOUT);
await clock.tickAsync(CONNECTION_ERROR_TIMEOUT);
consumer.stop();

sandbox.assert.calledTwice(errorListener);
Expand All @@ -457,7 +457,28 @@ describe("Consumer", () => {
consumer.on("error", errorListener);

consumer.start();
await clock.tickAsync(AUTHENTICATION_ERROR_TIMEOUT);
await clock.tickAsync(CONNECTION_ERROR_TIMEOUT);
consumer.stop();

sandbox.assert.calledTwice(errorListener);
sandbox.assert.calledTwice(sqs.send);
sandbox.assert.calledWithMatch(sqs.send.firstCall, mockReceiveMessage);
sandbox.assert.calledWithMatch(sqs.send.secondCall, mockReceiveMessage);
});

it("waits before repolling when a connection error occurs", async () => {
const unknownEndpointErr = {
name: "SQSError",
code: "SQSError",
message:
"SQS receive message failed: getaddrinfo ENOTFOUND sqs.eu-west-1.amazonaws.com",
};
sqs.send.withArgs(mockReceiveMessage).rejects(unknownEndpointErr);
const errorListener = sandbox.stub();
consumer.on("error", errorListener);

consumer.start();
await clock.tickAsync(CONNECTION_ERROR_TIMEOUT);
consumer.stop();

sandbox.assert.calledTwice(errorListener);
Expand All @@ -476,7 +497,7 @@ describe("Consumer", () => {
consumer.on("error", errorListener);

consumer.start();
await clock.tickAsync(AUTHENTICATION_ERROR_TIMEOUT);
await clock.tickAsync(CONNECTION_ERROR_TIMEOUT);
consumer.stop();

sandbox.assert.calledTwice(errorListener);
Expand All @@ -495,7 +516,7 @@ describe("Consumer", () => {
consumer.on("error", errorListener);

consumer.start();
await clock.tickAsync(AUTHENTICATION_ERROR_TIMEOUT);
await clock.tickAsync(CONNECTION_ERROR_TIMEOUT);
consumer.stop();

sandbox.assert.calledTwice(errorListener);
Expand All @@ -510,7 +531,7 @@ describe("Consumer", () => {
region: REGION,
handleMessage,
sqs,
authenticationErrorTimeout: AUTHENTICATION_ERROR_TIMEOUT,
connectionErrorTimeout: CONNECTION_ERROR_TIMEOUT,
pollingWaitTimeMs: POLLING_TIMEOUT,
});

Expand Down Expand Up @@ -560,7 +581,7 @@ describe("Consumer", () => {
region: REGION,
handleMessage,
sqs,
authenticationErrorTimeout: AUTHENTICATION_ERROR_TIMEOUT,
connectionErrorTimeout: CONNECTION_ERROR_TIMEOUT,
preReceiveMessageCallback: preReceiveMessageCallbackStub,
postReceiveMessageCallback: postReceiveMessageCallbackStub,
});
Expand Down Expand Up @@ -596,7 +617,7 @@ describe("Consumer", () => {
region: REGION,
handleMessage,
sqs,
authenticationErrorTimeout: AUTHENTICATION_ERROR_TIMEOUT,
connectionErrorTimeout: CONNECTION_ERROR_TIMEOUT,
shouldDeleteMessages: false,
});

Expand Down Expand Up @@ -701,7 +722,7 @@ describe("Consumer", () => {
AttributeNames: [],
MessageAttributeNames: ["attribute-1", "attribute-2"],
MaxNumberOfMessages: 3,
WaitTimeSeconds: AUTHENTICATION_ERROR_TIMEOUT,
WaitTimeSeconds: CONNECTION_ERROR_TIMEOUT,
VisibilityTimeout: undefined,
}),
);
Expand Down Expand Up @@ -745,7 +766,7 @@ describe("Consumer", () => {
AttributeNames: ["ApproximateReceiveCount"],
MessageAttributeNames: [],
MaxNumberOfMessages: 1,
WaitTimeSeconds: AUTHENTICATION_ERROR_TIMEOUT,
WaitTimeSeconds: CONNECTION_ERROR_TIMEOUT,
VisibilityTimeout: undefined,
}),
);
Expand Down Expand Up @@ -881,7 +902,7 @@ describe("Consumer", () => {
},
batchSize: 2,
sqs,
authenticationErrorTimeout: AUTHENTICATION_ERROR_TIMEOUT,
connectionErrorTimeout: CONNECTION_ERROR_TIMEOUT,
});

consumer.start();
Expand Down Expand Up @@ -917,7 +938,7 @@ describe("Consumer", () => {
},
batchSize: 2,
sqs,
authenticationErrorTimeout: AUTHENTICATION_ERROR_TIMEOUT,
connectionErrorTimeout: CONNECTION_ERROR_TIMEOUT,
});

consumer.start();
Expand All @@ -943,7 +964,7 @@ describe("Consumer", () => {
},
batchSize: 2,
sqs,
authenticationErrorTimeout: AUTHENTICATION_ERROR_TIMEOUT,
connectionErrorTimeout: CONNECTION_ERROR_TIMEOUT,
});

consumer.start();
Expand Down Expand Up @@ -1525,7 +1546,7 @@ describe("Consumer", () => {
handleMessage,
sqs,
pollingCompleteWaitTimeMs: 5000,
authenticationErrorTimeout: AUTHENTICATION_ERROR_TIMEOUT,
connectionErrorTimeout: CONNECTION_ERROR_TIMEOUT,
});

consumer.on("stopped", handleStop);
Expand Down Expand Up @@ -1579,7 +1600,7 @@ describe("Consumer", () => {
handleMessage,
sqs,
pollingCompleteWaitTimeMs: 500,
authenticationErrorTimeout: AUTHENTICATION_ERROR_TIMEOUT,
connectionErrorTimeout: CONNECTION_ERROR_TIMEOUT,
});

consumer.on("stopped", handleStop);
Expand Down
Loading