fix: address race condition between asset publishing and batch announcement (#728)

# Description
This PR does the following:
- overrides the retry limit for pre-existing batchAnnouncement jobs from
1 to 3
- traps failure to retrieve IPFS info of a pinned resource in the batch
queue processor and throws an error (prevents erroneous queueing of
batch jobs that end up looking like onChain content jobs)
- adds an initial delay of 3s to batchAnnouncement jobs to allow time
for asset upload to ipfs

Author: JoeCap08055

apps/content-publishing-api/src/api.service.ts

@@ -97,6 +97,8 @@ export class ApiService {
     data.id = this.calculateJobId(data);
     const job = await this.batchAnnouncerQueue.add(`Batch Request Job - ${data.id}`, data, {
       jobId: data.id,
+      attempts: 3,
+      delay: 3000,
       removeOnFail: false,
       removeOnComplete: 2000,
     }); // TODO: should come from queue configs

apps/content-publishing-worker/src/batch_announcer/batch.announcer.ts

@@ -71,7 +71,14 @@ export class BatchAnnouncer {
 
   public async announceExistingBatch(batch: IBatchFile): Promise<IPublisherJob> {
     // Get previously uploaded file from IPFS
-    const { Key: cid, Size: size } = await this.ipfsService.getInfo(batch.cid);
+    this.logger.log(`Getting info from IPFS for ${batch.cid}`);
+    const { Key: cid, Size: size, Message: msg, Type: msgType } = await this.ipfsService.getInfo(batch.cid);
+    if (msgType === 'error' || !cid) {
+      this.logger.error(`Unable to confirm batch file existence in IPFS: ${msg}`);
+      throw new Error(`Unable to confirm batch file existence in IPFS: ${msg}`);
+    }
+
+    this.logger.debug(`Got info from IPFS: cid=${cid}, size=${size}`);
 
     const ipfsUrl = await this.formIpfsUrl(cid);
     const response = { id: batch.cid, schemaId: batch.schemaId, data: { cid, payloadLength: size } };

libs/storage/src/ipfs/ipfs.service.ts

@@ -61,7 +61,7 @@ export class IpfsService {
 
   public async getInfo(cid: string, checkExistence = true): Promise<IpfsBlockStatResponse> {
     if (checkExistence && !(await this.isPinned(cid))) {
-      return Promise.resolve({ Message: 'Requested resource does not exist', Type: 'error' });
+      return { Message: 'Requested resource does not exist', Type: 'error' };
     }
 
     const ipfsGet = `${this.config.ipfsEndpoint}/api/v0/block/stat?arg=${cid}`;
@@ -74,6 +74,7 @@ export class IpfsService {
 
     const headers = { Accept: '*/*', Connection: 'keep-alive', authorization: ipfsAuth };
 
+    this.logger.debug(`Requesting IPFS stats from ${ipfsGet}`);
     const response = await axios.post(ipfsGet, null, { headers, responseType: 'json' });
     this.logger.debug(`IPFS response: ${JSON.stringify(response.data)}`);
     return response.data as IpfsBlockStatResponse;
@@ -96,8 +97,9 @@ export class IpfsService {
       authorization: ipfsAuth,
     };
 
+    this.logger.debug(`Requesting pin info from IPFS for ${ipfsGet}`);
     const response = await axios.post(ipfsGet, null, { headers, responseType: 'json' }).catch((error) => {
-      // when pid does not exist this call returns 500 which is not great
+      // when pin does not exist this call returns 500 which is not great
       if (error.response && error.response.status !== 500) {
         this.logger.error(error.toJSON());
       }