fix: handle pg disconnections and transaction management (#92)

* chore: upgrade postgres.js

* chore: add auth failed to connection errors

* fix: pg transaction and conn error handling

* fix: error messages

* test: job pg errors

* test: queue loop pg error

Author: rafaelcr

package-lock.json

@@ -39,6 +39,8 @@ export function isPgConnectionError(error: any): string | false {
       return 'Postgres connection failed due to a DNS lookup error';
     } else if (msg.includes('terminating connection due to administrator command')) {
       return 'Postgres connection closed due to administrator command';
+    } else if (msg.includes('password authentication failed')) {
+      return 'Postgres authentication failed';
     }
   }
   return false;

src/pg/postgres-tools/errors.ts

@@ -85,19 +85,22 @@ export class JobQueue {
     ) {
       return;
     }
-    this.jobIds.add(job.id);
     await this.db.updateJobStatus({ id: job.id, status: DbJobStatus.queued });
+    this.jobIds.add(job.id);
     void this.queue.add(async () => {
-      if (this.isRunning) {
-        if (job.token_id) {
-          await new ProcessTokenJob({ db: this.db, job: job }).work();
-        } else if (job.smart_contract_id) {
-          await new ProcessSmartContractJob({ db: this.db, apiDb: this.apiDb, job: job }).work();
+      try {
+        if (this.isRunning) {
+          if (job.token_id) {
+            await new ProcessTokenJob({ db: this.db, job: job }).work();
+          } else if (job.smart_contract_id) {
+            await new ProcessSmartContractJob({ db: this.db, apiDb: this.apiDb, job: job }).work();
+          }
+        } else {
+          logger.info(`JobQueue cancelling job ${job.id}, queue is now closed`);
         }
-      } else {
-        logger.info(`JobQueue cancelling job ${job.id}, queue is now closed`);
+      } finally {
+        this.jobIds.delete(job.id);
       }
-      this.jobIds.delete(job.id);
     });
   }
 

src/token-processor/queue/job-queue.ts

@@ -38,16 +38,13 @@ export abstract class Job {
     let finishedWithError = false;
     const sw = stopwatch();
 
-    // This sql transaction will catch any and all errors that are generated while processing the
-    // job. Each of them were previously tagged as retryable or not retryable so we'll make a
-    // decision here about what to do in each case. If we choose to retry, this queue entry will
-    // simply not be marked as `processed = true` so it can be picked up by the queue at a later
-    // time.
+    // This block will catch any and all errors that are generated while processing the job. Each of
+    // them were previously tagged as retryable or not retryable so we'll make a decision here about
+    // what to do in each case. If we choose to retry, this queue entry will simply not be marked as
+    // `processed = true` so it can be picked up by the queue at a later time.
     try {
-      await this.db.sqlWriteTransaction(async sql => {
-        await this.handler();
-        processingFinished = true;
-      });
+      await this.handler();
+      processingFinished = true;
     } catch (error) {
       if (error instanceof RetryableJobError) {
         const retries = await this.db.increaseJobRetryCount({ id: this.job.id });
@@ -59,7 +56,7 @@ export abstract class Job {
             error,
             `Job ${this.description()} recoverable error after ${sw.getElapsed()}ms, trying again later`
           );
-          await this.db.updateJobStatus({ id: this.job.id, status: DbJobStatus.pending });
+          await this.updateStatus(DbJobStatus.pending);
         } else {
           logger.warn(error, `Job ${this.description()} max retries reached, giving up`);
           processingFinished = true;
@@ -74,12 +71,20 @@ export abstract class Job {
     } finally {
       if (processingFinished) {
         const status = finishedWithError ? DbJobStatus.failed : DbJobStatus.done;
-        await this.db.updateJobStatus({
-          id: this.job.id,
-          status: status,
-        });
-        logger.info(`Job ${this.description()} ${status} in ${sw.getElapsed()}ms`);
+        if (await this.updateStatus(status)) {
+          logger.info(`Job ${this.description()} ${status} in ${sw.getElapsed()}ms`);
+        }
       }
     }
   }
+
+  private async updateStatus(status: DbJobStatus): Promise<boolean> {
+    try {
+      await this.db.updateJobStatus({ id: this.job.id, status: status });
+      return true;
+    } catch (error) {
+      logger.error(`Job ${this.description()} could not update status to ${status}: ${error}`);
+      return false;
+    }
+  }
 }

src/token-processor/queue/job/job.ts

@@ -105,14 +105,16 @@ export class ProcessSmartContractJob extends Job {
       );
       return;
     }
-    await this.db.updateSmartContractTokenCount({ id: contract.id, count: tokenCount });
-    logger.info(
-      `ProcessSmartContractJob enqueueing ${tokenCount} tokens for ${this.description()}`
-    );
-    await this.db.insertAndEnqueueSequentialTokens({
-      smart_contract_id: contract.id,
-      token_count: tokenCount,
-      type: dbSipNumberToDbTokenType(contract.sip),
+    await this.db.sqlWriteTransaction(async sql => {
+      logger.info(
+        `ProcessSmartContractJob enqueueing ${tokenCount} tokens for ${this.description()}`
+      );
+      await this.db.updateSmartContractTokenCount({ id: contract.id, count: tokenCount });
+      await this.db.insertAndEnqueueSequentialTokens({
+        smart_contract_id: contract.id,
+        token_count: tokenCount,
+        type: dbSipNumberToDbTokenType(contract.sip),
+      });
     });
   }
 }

src/token-processor/queue/job/process-smart-contract-job.ts

@@ -30,17 +30,21 @@ export class ProcessTokenJob extends Job {
   private contract?: DbSmartContract;
 
   async handler(): Promise<void> {
-    if (!this.job.token_id) {
+    const tokenId = this.job.token_id;
+    if (!tokenId) {
       return;
     }
-    const token = await this.db.getToken({ id: this.job.token_id });
-    if (!token) {
-      throw Error(`ProcessTokenJob token not found with id ${this.job.token_id}`);
-    }
-    const contract = await this.db.getSmartContract({ id: token.smart_contract_id });
-    if (!contract) {
-      throw Error(`ProcessTokenJob contract not found with id ${token.smart_contract_id}`);
-    }
+    const [token, contract] = await this.db.sqlTransaction(async sql => {
+      const token = await this.db.getToken({ id: tokenId });
+      if (!token) {
+        throw Error(`ProcessTokenJob token not found with id ${tokenId}`);
+      }
+      const contract = await this.db.getSmartContract({ id: token.smart_contract_id });
+      if (!contract) {
+        throw Error(`ProcessTokenJob contract not found with id ${token.smart_contract_id}`);
+      }
+      return [token, contract];
+    });
     this.token = token;
     this.contract = contract;
 

src/token-processor/queue/job/process-token-job.ts

@@ -5,6 +5,7 @@ import { DbJob, DbJobStatus, DbSipNumber, DbSmartContractInsert } from '../src/p
 import { JobQueue } from '../src/token-processor/queue/job-queue';
 import { cycleMigrations } from '../src/pg/migrations';
 import { PgBlockchainApiStore } from '../src/pg/blockchain-api/pg-blockchain-api-store';
+import { MockPgBlockchainApiStore } from './helpers';
 
 class TestJobQueue extends JobQueue {
   constructor(args: { db: PgStore; apiDb: PgBlockchainApiStore }) {
@@ -115,4 +116,27 @@ describe('JobQueue', () => {
     expect((await db.getJob({ id: job2.id }))?.status).toBe('queued');
     expect((await db.getJob({ id: job3.id }))?.status).toBe('queued');
   });
+
+  test('pg connection errors are not re-thrown', async () => {
+    const values1: DbSmartContractInsert = {
+      principal: 'ABCD.test-ft',
+      sip: DbSipNumber.sip010,
+      abi: '"some"',
+      tx_id: '0x123456',
+      block_height: 1,
+    };
+    await db.insertAndEnqueueSmartContract({ values: values1 });
+
+    const queue = new JobQueue({ db, apiDb: new MockPgBlockchainApiStore() });
+    const sleep = (time: number) => {
+      return new Promise(resolve => setTimeout(resolve, time));
+    };
+
+    // Close DB and start the queue. If the error is not handled correctly, the test will fail.
+    await db.close();
+    queue.start();
+    // Wait 2 seconds and kill the queue.
+    await sleep(2000);
+    await queue.close();
+  });
 });

tests/job-queue.test.ts

@@ -9,11 +9,20 @@ class TestRetryableJob extends Job {
   description(): string {
     return 'test';
   }
-  protected handler(): Promise<void> {
+  handler(): Promise<void> {
     throw new RetryableJobError('test');
   }
 }
 
+class TestDbJob extends Job {
+  description(): string {
+    return 'test';
+  }
+  async handler(): Promise<void> {
+    await this.db.sql<{ version: string }[]>`SELECT version()`;
+  }
+}
+
 describe('Job', () => {
   let db: PgStore;
   let dbJob: DbJob;
@@ -70,4 +79,10 @@ describe('Job', () => {
     expect(jobs1[0].retry_count).toBe(1);
     expect(jobs1[0].status).toBe('pending');
   });
+
+  test('db errors are not re-thrown', async () => {
+    await db.close();
+    const job = new TestDbJob({ db, job: dbJob });
+    await expect(job.work()).resolves.not.toThrow();
+  });
 });