Fix intermittent test failure

Author: jwulf

.github/workflows/npm-publish.yml

@@ -1,8 +1,8 @@
 name: npm-publish
 on:
   push:
-    tags:
-      - "v*"
+    branches:
+      - "master"
 jobs:
   test:
     name: test

DEVELOP.md

@@ -0,0 +1,7 @@
+# Development
+
+## Publishing a new NPM Package
+
+The NPM package publishing is handled by a GitHub Workflows using the [publish-to-npm](https://github.com/marketplace/actions/publish-to-npm) action.
+
+To publish a new package, update the `package.json` package version and push a commit with the commit message "Release x.y.z" matching the package version.

package-lock.json

@@ -30,7 +30,7 @@
 		"build": "tsc --build src/tsconfig.json",
 		"watch": "tsc --build src/tsconfig.json -w",
 		"prepare": "tsc --build src/tsconfig.json",
-		"test": "jest --runInBand --detectOpenHandles --testPathIgnorePatterns integration local-integration disconnection",
+		"test": "jest --detectOpenHandles --testPathIgnorePatterns integration local-integration disconnection",
 		"test:integration": "jest --runInBand --testPathIgnorePatterns disconnection --detectOpenHandles --verbose true",
 		"test:local": "jest --runInBand --verbose true --detectOpenHandles local-integration",
 		"test:disconnect": "jest --runInBand --verbose true --detectOpenHandles disconnection",
@@ -65,7 +65,7 @@
 		"chalk": "^2.4.2",
 		"console-stamp": "^0.2.7",
 		"dayjs": "^1.8.15",
-		"debug": "^4.1.1",
+		"debug": "^4.2.0",
 		"fast-xml-parser": "^3.12.12",
 		"fp-ts": "^2.5.1",
 		"got": "^9.6.0",

package.json

@@ -6,7 +6,7 @@ process.env.ZEEBE_NODE_LOG_LEVEL = process.env.ZEEBE_NODE_LOG_LEVEL || 'NONE'
 
 const ZEEBE_DOCKER_TAG = '0.24.2'
 
-jest.setTimeout(120000)
+jest.setTimeout(900000)
 
 let container
 let worker
@@ -16,14 +16,19 @@ afterEach(async () => {
 	await worker?.close()
 })
 
+function log(msg) {
+	// tslint:disable-next-line: no-console
+	console.log(new Date().toString(), msg) // @DEBUG
+}
+
 test('reconnects after a pod reschedule', async done => {
 	let readyCount = 0
 	let errorCount = 0
 	const delay = timeout =>
 		new Promise(res => setTimeout(() => res(), timeout))
 
 	// tslint:disable-next-line: no-console
-	console.log('##### Starting container') // @DEBUG
+	log('##### Starting container') // @DEBUG
 
 	container = await new GenericContainer(
 		'camunda/zeebe',
@@ -38,10 +43,18 @@ test('reconnects after a pod reschedule', async done => {
 	await delay(10000)
 
 	const zbc = new ZBClient(`localhost`)
+	// tslint:disable-next-line: no-console
+	log('##### Deploying workflow') // @DEBUG
+
 	await zbc.deployWorkflow('./src/__tests__/testdata/disconnection.bpmn')
 	worker = zbc
 		.createWorker({
+			longPoll: 10000,
+			pollInterval: 300,
 			taskHandler: (_, complete) => {
+				// tslint:disable-next-line: no-console
+				log('##### Executing task handler') // @DEBUG
+
 				complete.success()
 			},
 			taskType: 'disconnection-task',
@@ -54,7 +67,7 @@ test('reconnects after a pod reschedule', async done => {
 		})
 
 	// tslint:disable-next-line: no-console
-	console.log('##### Deploying workflow') // @DEBUG
+	log('##### Starting workflow') // @DEBUG
 
 	const wf = await zbc.createWorkflowInstanceWithResult({
 		bpmnProcessId: 'disconnection',
@@ -64,15 +77,18 @@ test('reconnects after a pod reschedule', async done => {
 	expect(wf.bpmnProcessId).toBeTruthy()
 
 	// tslint:disable-next-line: no-console
-	console.log('##### Stopping container...') // @DEBUG
+	log('##### Workflow finished') // @DEBUG
+
+	// tslint:disable-next-line: no-console
+	log('##### Stopping container...') // @DEBUG
 
 	await container.stop()
 
 	// tslint:disable-next-line: no-console
-	console.log('##### Container stopped.') // @DEBUG
+	log('##### Container stopped.') // @DEBUG
 
 	// tslint:disable-next-line: no-console
-	console.log('##### Starting container....') // @DEBUG
+	log('##### Starting container....') // @DEBUG
 
 	container = await new GenericContainer(
 		'camunda/zeebe',
@@ -81,15 +97,27 @@ test('reconnects after a pod reschedule', async done => {
 		26500
 	)
 		.withExposedPorts(26500)
+		.withEnv('ZEEBE_LOG_LEVEL', 'trace')
 		.withWaitStrategy(Wait.forLogMessage('Bootstrap Broker-0 succeeded.'))
 		.start()
 
 	// tslint:disable-next-line: no-console
-	// console.log('##### Container started.') // @DEBUG
+	log('##### Container started.') // @DEBUG
 
 	await delay(10000)
+
+	// tslint:disable-next-line: no-console
+	log('##### Deploying workflow 2') // @DEBUG
 	await zbc.deployWorkflow('./src/__tests__/testdata/disconnection.bpmn')
 
+	// tslint:disable-next-line: no-console
+	// console.log('Workflow 2 deployed', _) // @DEBUG
+
+	await delay(15000)
+
+	// tslint:disable-next-line: no-console
+	log('##### Starting workflow 2') // @DEBUG
+
 	const wf1 = await zbc.createWorkflowInstanceWithResult('disconnection', {})
 	expect(wf1.bpmnProcessId).toBeTruthy()
 	await worker.close()

src/__tests__/disconnection/disconnect.spec.ts

@@ -18,10 +18,10 @@ test(`Calls the onConnectionError handler if there is no broker and eagerConnect
 	}, 7000)
 })
 
-test(`Sets connected:undefined if there is no broker and no setting of eagerConnection`, async done => {
+test(`Sets connected:false if there is no broker and no setting of eagerConnection`, async done => {
 	const zbc2 = new ZBClient('localtoast: 267890') // Broker doesn't exist!!!
 	setTimeout(async () => {
-		expect(zbc2.connected).toBe(undefined)
+		expect(zbc2.connected).toBe(false)
 		await zbc2.close()
 		done()
 	}, 5000)

src/__tests__/integration/Client-ConnectionError.spec.ts

@@ -7,14 +7,9 @@ process.env.ZEEBE_NODE_LOG_LEVEL = process.env.ZEEBE_NODE_LOG_LEVEL || 'NONE'
 let zbcLongPoll
 
 afterAll(async () => {
-	const zbc = new ZBClient()
 	await zbcLongPoll.close()
-	await zbc.close()
 })
 
-/**
- * This test is currently disabled
- */
 test('Does long poll by default', async done => {
 	jest.setTimeout(40000)
 	zbcLongPoll = new ZBClient({

src/__tests__/integration/Worker-LongPoll.spec.ts

@@ -48,7 +48,7 @@ test(`Does set connected: true if there is a broker and eagerConnection: true`,
 test(`Does not set connected: true if there is a broker and eagerConnection: false`, done => {
 	const zbc2 = new ZBClient()
 	setTimeout(async () => {
-		expect(zbc2.connected).toBe(undefined)
+		expect(zbc2.connected).toBe(false)
 		await zbc2.close()
 		done()
 	}, 7000)

src/__tests__/integration/Worker-onReady.spec.ts

@@ -65,7 +65,7 @@ test(`Calls ZBClient onConnectionError when there no broker, for the client and
 	})
 	setTimeout(() => {
 		zbc2.close()
-		expect(calledD).toBe(1)
+		expect(calledD).toBe(2)
 		done()
 	}, 10000)
 })
@@ -82,7 +82,7 @@ test(`Debounces onConnectionError`, async done => {
 	})
 	setTimeout(() => {
 		zbc2.close()
-		expect(called).toBe(1) // toBeLessThanOrEqual(1)
+		expect(called).toBe(2) // toBeLessThanOrEqual(1)
 		done()
 	}, 15000)
 })
@@ -108,6 +108,7 @@ test(`Does not call the onConnectionError handler if there is a business error`,
 			calledF++
 		},
 	})
+
 	zbc2.createWorkflowInstance(wf, {}).catch(() => {
 		wf = 'throw error away'
 	})

src/__tests__/local-integration/OnConnectionError.spec.ts

@@ -6,6 +6,7 @@ import {
 	status as statusJS,
 } from '@grpc/grpc-js'
 import { loadSync, Options, PackageDefinition } from '@grpc/proto-loader'
+import * as _debug from 'debug'
 import { EventEmitter } from 'events'
 import {
 	Client,
@@ -20,6 +21,8 @@ import { BasicAuthConfig } from './interfaces'
 import { Loglevel } from './interfaces-published-contract'
 import { OAuthProvider } from './OAuthProvider'
 
+const debug = _debug.default('grpc')
+
 const useJS = process.env.ZEEBE_NODE_PUREJS?.toUpperCase() === 'TRUE'
 const credentials = useJS ? credentialsJS : credentialsC
 const InterceptingCall = useJS ? InterceptingCallJS : InterceptingCallC
@@ -222,12 +225,16 @@ export class GrpcClient extends EventEmitter {
 			 * The maximum time between subsequent connection attempts,
 			 * in ms
 			 */
-			'grpc.max_reconnect_backoff_ms': 30000,
+			'grpc.max_reconnect_backoff_ms': 10000,
 			/**
 			 * The minimum time between subsequent connection attempts,
-			 * in ms
+			 * in ms. Default is 1000ms, but this can cause an SSL Handshake failure.
+			 * This causes an intermittent failure in the Worker-LongPoll test when run
+			 * against Camunda Cloud.
+			 * Raised to 5000ms.
+			 * See: https://github.com/grpc/grpc/issues/8382#issuecomment-259482949
 			 */
-			'grpc.min_reconnect_backoff_ms': 1000,
+			'grpc.min_reconnect_backoff_ms': 5000,
 			/**
 			 * After a duration of this time the client/server
 			 * pings its peer to see if the transport is still alive.
@@ -240,7 +247,7 @@ export class GrpcClient extends EventEmitter {
 			 * not receive the ping ack, it will close the
 			 * transport. Int valued, milliseconds.
 			 */
-			'grpc.keepalive_timeout_ms': 60000,
+			'grpc.keepalive_timeout_ms': 120000,
 			'grpc.http2.min_time_between_pings_ms': 60000,
 			/**
 			 * Minimum allowed time between a server receiving
@@ -267,6 +274,12 @@ export class GrpcClient extends EventEmitter {
 		})
 		this.listNameMethods = []
 
+		this.client.waitForReady(10000, error =>
+			error
+				? this.emit(MiddlewareSignals.Event.Error, error)
+				: this.emit(MiddlewareSignals.Event.Ready)
+		)
+
 		for (const key in listMethods) {
 			if (listMethods[key]) {
 				const methodName = listMethods[key].originalName as string
@@ -286,6 +299,7 @@ export class GrpcClient extends EventEmitter {
 					)
 					try {
 						const metadata = await this.getAuthToken()
+
 						stream = this.client[methodName](
 							timeNormalisedRequest,
 							metadata
@@ -304,6 +318,19 @@ export class GrpcClient extends EventEmitter {
 							),
 						}
 					}
+
+					// This deals with the case where during a broker restart the call returns a stream
+					// but that stream is not a legit Gateway activation. In that case, the Gateway will
+					// never time out or close the stream. So we have to manage that case.
+					const clientsideTimeoutDuration =
+						Duration.milliseconds.from(this.longPoll!) + 1000
+					const clientSideTimeout = setTimeout(() => {
+						debug(
+							`Triggered client-side timeout after ${clientsideTimeoutDuration}ms`
+						)
+						stream.emit('end')
+					}, clientsideTimeoutDuration)
+
 					/**
 					 * Once this gets attached here, it is attached to *all* calls
 					 * This is an issue if you do a sync call like cancelWorkflowSync
@@ -312,6 +339,8 @@ export class GrpcClient extends EventEmitter {
 					 * streaming calls, and each worker, which only does streaming calls
 					 */
 					stream.on('error', (error: GrpcStreamError) => {
+						clearTimeout(clientSideTimeout)
+						debug(`Error`, error)
 						this.emit(MiddlewareSignals.Event.Error)
 						if (error.message.includes('14 UNAVAILABLE')) {
 							this.emit(
@@ -341,6 +370,8 @@ export class GrpcClient extends EventEmitter {
 							`gRPC Status event: ${JSON.stringify(s)}`
 						)
 					)
+					stream.on('end', () => clearTimeout(clientSideTimeout))
+
 					return stream
 				}
 

src/lib/GrpcClient.ts

@@ -1,5 +1,6 @@
 import { ClientReadableStreamImpl } from '@grpc/grpc-js/build/src/call'
 import { Chalk } from 'chalk'
+import * as _debug from 'debug'
 import { EventEmitter } from 'events'
 import { Duration, MaybeTimeDuration } from 'typed-duration'
 import * as uuid from 'uuid'
@@ -11,6 +12,8 @@ import { ActivateJobsRequest, ActivateJobsResponse } from './interfaces-grpc'
 import { ZBClientOptions } from './interfaces-published-contract'
 import { TypedEmitter } from './TypedEmitter'
 
+const debug = _debug.default('worker')
+
 const MIN_ACTIVE_JOBS_RATIO_BEFORE_ACTIVATING_JOBS = 0.3
 
 const CapacityEvent = {
@@ -99,6 +102,7 @@ export class ZBWorkerBase<
 	private maxActiveJobsBeforeReactivation: number
 	private pollInterval: MaybeTimeDuration
 	private pollLoop: NodeJS.Timeout
+	private pollMutex: boolean = false
 
 	constructor({
 		grpcClient,
@@ -369,7 +373,7 @@ export class ZBWorkerBase<
 	}
 
 	private async poll() {
-		if (this.closePromise) {
+		if (this.closePromise || this.pollMutex) {
 			return
 		}
 		const isOverCapacity =
@@ -383,6 +387,8 @@ export class ZBWorkerBase<
 		if (this.jobStream) {
 			return
 		}
+		this.pollMutex = true
+		debug('Polling...')
 		this.logger.logDebug('Activating Jobs...')
 		const id = uuid.v4()
 		const jobStream = await this.activateJobs(id)
@@ -421,6 +427,7 @@ export class ZBWorkerBase<
 		if (jobStream.error) {
 			this.logger.logError({ id, error: jobStream.error.message })
 		}
+		this.pollMutex = false
 	}
 
 	private async activateJobs(id: string) {
@@ -435,6 +442,7 @@ export class ZBWorkerBase<
 		if (this.debugMode) {
 			this.logger.logDebug(`Activating Jobs...`)
 		}
+		debug('Activating Jobs')
 		let stream: any
 
 		const amount = this.maxJobsToActivate - this.activeJobs
@@ -453,6 +461,11 @@ export class ZBWorkerBase<
 				requestTimeout
 			)}, job timeout: ${Duration.value.of(this.timeout)}`
 		)
+		debug(
+			`Requesting ${amount} jobs on [${id}] with requestTimeout ${Duration.value.of(
+				requestTimeout
+			)}, job timeout: ${Duration.value.of(this.timeout)}`
+		)
 
 		try {
 			stream = await this.grpcClient.activateJobsStream(