9604164031

Committed 20 Jun 2024 08:47PM UTC coverage: 15.01% (+0.004%) from 15.006%

Build # 9604164031

Build Type

Pull #8200

github

Committed by

web-flow

Commit Message

Merge e4ceea889 into f096fad97

Pull Request Pull Request #8200: Fix relayer notify bug

Run Details

3722 of 26289 branches covered (14.16%)

Branch coverage included in aggregate %.

1 of 5 new or added lines in 2 files covered. (20.0%)

122 existing lines in 23 files now uncovered.

5154 of 32843 relevant lines covered (15.69%)

5.02 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

41.94

/packages/commonwealth/server/workers/messageRelayer/pgListener.ts

import { dispose, logger, stats } from '@hicommonwealth/core';
import { delay } from '@hicommonwealth/shared';
import pg from 'pg';
import { fileURLToPath } from 'url';
import { incrementNumUnrelayedEvents } from './relayForever';

const __filename = fileURLToPath(import.meta.url);
const log = logger(__filename);
const OUTBOX_CHANNEL = 'outbox_channel';
let retryCount = 0;
const maxRetries = 5;
let connected = false;

async function connectListener(client: pg.Client) {
  try {
    await client.query(`LISTEN "${OUTBOX_CHANNEL}";`);
  } catch (err) {
    log.fatal('Failed to setup Postgres listener. Exiting...', err);
    await dispose()('ERROR', true);
  }
}

async function reconnect(client: pg.Client) {
  if (retryCount < maxRetries) {
    // Exponential backoff strategy for reconnection attempts
    const timeout = Math.pow(2, retryCount) * 1000;
    log.warn(`Attempting to reconnect in ${timeout / 1000} seconds...`);
    retryCount++;

    try {
      if (!connected) {
        await client.connect();
        connected = true;
      }
    } catch (err) {
      log.error('Subscriber failed to reconnect', err);
      await delay(timeout);
      return reconnect(client);
    }
  } else {
    log.fatal('Max retry attempts reached. Exiting...');
    try {
      await client.end();
    } catch (e) {
      log.error('Failed to close pg client', e);
    }
    await dispose()('ERROR', true);
  }
}

export async function setupListener(): Promise<pg.Client> {
  log.info('Setting up listener...');
  const { config } = await import('@hicommonwealth/model');
  const client = new pg.Client({
    connectionString: config.DB.URI,
    ssl: ['test', 'development'].includes(config.NODE_ENV)
      ? false
      : { rejectUnauthorized: false },
  });

  client.on('notification', (payload) => {
    log.info('Notification received', { payload });
    incrementNumUnrelayedEvents(1);
    stats().increment('messageRelayerNotificationReceived');
  });

  client.on('error', (err: Error) => {
    log.error(
      'PG subscriber encountered an error. Attempting to reconnect...',
      err,
    );
    connected = false;
    reconnect(client)
      .then(() => connectListener(client))
      .catch(async (e) => {
        log.fatal('Failed to reconnect after pg error', e);
        await dispose()('ERROR', true);
      });
  });

  client.on('end', () => {
    connected = false;
  });

  // client will either connect/reconnect or the process will exit
  try {
    if (!connected) {
      await client.connect();
    }
  } catch (err) {
    log.error('Subscriber failed to connect to Postgres', err);
    await reconnect(client);
  }

  connected = true;
  await connectListener(client);

  log.info('Listener ready');
  return client;
}

1	import { dispose, logger, stats } from '@hicommonwealth/core';
2	import { delay } from '@hicommonwealth/shared';
3	import pg from 'pg';
4	import { fileURLToPath } from 'url';
5	import { incrementNumUnrelayedEvents } from './relayForever';
6
7	const __filename = fileURLToPath(import.meta.url);
8	const log = logger(__filename);	2✔
9	const OUTBOX_CHANNEL = 'outbox_channel';	2✔
10	let retryCount = 0;	2✔
11	const maxRetries = 5;	2✔
12	let connected = false;	2✔
13		2✔
14	async function connectListener(client: pg.Client) {
15	try {
16	await client.query(`LISTEN "${OUTBOX_CHANNEL}";`);	2✔
17	} catch (err) {	2✔
18	log.fatal('Failed to setup Postgres listener. Exiting...', err);
19	await dispose()('ERROR', true);	×
20	}
21	}
22
23	async function reconnect(client: pg.Client) {
24	if (retryCount < maxRetries) {	×
25	// Exponential backoff strategy for reconnection attempts	×
26	const timeout = Math.pow(2, retryCount) * 1000;
27	log.warn(`Attempting to reconnect in ${timeout / 1000} seconds...`);	×
28	retryCount++;	×
29
30	try {
31	if (!connected) {	×
32	await client.connect();	×
33	connected = true;	×
34	}
35	} catch (err) {
36	log.error('Subscriber failed to reconnect', err);
37	await delay(timeout);	×
38	return reconnect(client);	×
39	}
40	} else {
41	log.fatal('Max retry attempts reached. Exiting...');
42	try {	×
43	await client.end();	×
44	} catch (e) {
45	log.error('Failed to close pg client', e);
46	}
47	await dispose()('ERROR', true);
48	}
49	}
50
51	export async function setupListener(): Promise<pg.Client> {
52	log.info('Setting up listener...');
53	const { config } = await import('@hicommonwealth/model');	2✔
54	const client = new pg.Client({	2✔
55	connectionString: config.DB.URI,	2✔
56	ssl: ['test', 'development'].includes(config.NODE_ENV)	×
57	? false	4!
58	: { rejectUnauthorized: false },
59	});
60
61	client.on('notification', (payload) => {
62	log.info('Notification received', { payload });	2✔
NEW 63	incrementNumUnrelayedEvents(1);	×
NEW 64	stats().increment('messageRelayerNotificationReceived');	×
65	});
66
67	client.on('error', (err: Error) => {
68	log.error(
69	'PG subscriber encountered an error. Attempting to reconnect...',
70	err,
71	);
NEW 72	connected = false;	×
NEW 73	reconnect(client)	×
74	.then(() => connectListener(client))
75	.catch(async (e) => {
76	log.fatal('Failed to reconnect after pg error', e);
77	await dispose()('ERROR', true);
78	});	2✔
79	});
80
81	client.on('end', () => {
82	connected = false;
83	});
84
85	// client will either connect/reconnect or the process will exit
86	try {
87	if (!connected) {	×
88	await client.connect();	×
89	}
90	} catch (err) {
91	log.error('Subscriber failed to connect to Postgres', err);
92	await reconnect(client);	2✔
93	}	2✔
94
95	connected = true;
96	await connectListener(client);
97		2✔
98	log.info('Listener ready');	2!
99	return client;	2✔
100	}

hicommonwealth / commonwealth / 9604164031

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous