Dealmaking bugfix4 (#156)

Various fixes and enhancements on dealmaking: csv support, force send support, deal tracking bug

Author: kernelogic

.gitignore

@@ -120,3 +120,4 @@ outDir
 nunitresults.xml
 config/cars/
 
+.DS_Store

README.md

@@ -326,8 +326,10 @@ Options:
   -c, --cron-schedule <cronschedule>                   Optional cron to send deals at interval. Use double quote to wrap the format containing spaces.
   -x, --cron-max-deals <cronmaxdeals>                  When cron schedule specified, limit the total number of deals across entire cron, per SP.
   -xp, --cron-max-pending-deals <cronmaxpendingdeals>  When cron schedule specified, limit the total number of pending deals determined by dealtracking service, per SP.
-  -l, --file-list-path <filelistpath>                  Absolute path to a txt file that will limit to replicate only from the list. Must be visible by deal replication worker.
+  -l, --file-list-path <filelistpath>                  Path to a txt file that will limit to replicate only from the list. Must be visible by deal replication worker.
   -n, --notes <notes>                                  Any notes or tag want to store along the replication request, for tracking purpose.
+  -csv, --output-csv <outputCsv>                       Print CSV to specified folder after done. Folder must exist on worker.
+  -f, --force                                          Force resend even if this pieceCID have been proposed / active by the provider. (default: false)
   -h, --help                                           display help for command
 ```
 

src/common/Datastore.ts

@@ -184,6 +184,8 @@ export default class Datastore {
       cronMaxPendingDeals: Schema.Types.Number,
       fileListPath: Schema.Types.String,
       notes: Schema.Types.String,
+      csvOutputDir: Schema.Types.String,
+      isForced: Schema.Types.Boolean,
       errorMessage: Schema.Types.String
     }, {
       timestamps: true

src/common/GenerateCsv.ts

@@ -0,0 +1,60 @@
+import path from 'path';
+import fs from 'fs-extra';
+import DealReplicationWorker from '../replication/DealReplicationWorker';
+import Datastore from './Datastore';
+import ObjectsToCsv from 'objects-to-csv';
+
+/**
+ * Export replication request to CSV
+ */
+export default class GenerateCsv {
+  public static async generate (id: string, outDir: string): Promise<string> {
+    let msg = '';
+    fs.mkdirpSync(outDir);
+    const replicationRequest = await Datastore.ReplicationRequestModel.findById(id);
+    if (replicationRequest) {
+      const providers = DealReplicationWorker.generateProvidersList(replicationRequest.storageProviders);
+      for (let j = 0; j < providers.length; j++) {
+        const provider = providers[j];
+        const deals = await Datastore.DealStateModel.find({
+          replicationRequestId: id,
+          provider: provider,
+          state: { $nin: ['slashed', 'error', 'expired', 'proposal_expired'] }
+        });
+        let urlPrefix = replicationRequest.urlPrefix;
+        if (!urlPrefix.endsWith('/')) {
+          urlPrefix += '/';
+        }
+
+        if (deals.length > 0) {
+          const csvRow = [];
+          for (let i = 0; i < deals.length; i++) {
+            const deal = deals[i];
+            csvRow.push({
+              miner_id: deal.provider,
+              deal_cid: deal.dealCid,
+              filename: `${deal.pieceCid}.car`,
+              data_cid: deal.dataCid,
+              piece_cid: deal.pieceCid,
+              start_epoch: deal.startEpoch,
+              full_url: `${urlPrefix}${deal.pieceCid}.car`
+            });
+          }
+          const csv = new ObjectsToCsv(csvRow);
+          let fileListFilename = '';
+          if (replicationRequest.fileListPath) {
+            fileListFilename += '_' + path.parse(replicationRequest.fileListPath).name;
+          }
+          const filename = path.join(outDir, `${provider}${fileListFilename}_${id}.csv`);
+          await csv.toDisk(filename);
+          msg += `CSV saved to ${filename}\n`;
+        } else {
+          msg += `No deal found to export in ${id}\n`;
+        }
+      }
+    } else {
+      msg = `Replication request not found ${id}`;
+    }
+    return msg;
+  }
+}

src/common/model/ReplicationRequest.ts

@@ -21,5 +21,7 @@ export default interface ReplicationRequest {
   cronMaxPendingDeals?: number, // per SP pending total with cron considered
   fileListPath?: string, // limit to replicate only from the list in a txt file
   notes?: string, // any notes or tag want to store along the replication request, for tracking purpose
+  csvOutputDir?: string, // folder to print CSV to, undefined to skip the CSV
+  isForced: boolean,
   errorMessage?: string
 }

src/deal-tracking/DealTrackingService.ts

@@ -141,61 +141,6 @@ export default class DealTrackingService extends BaseService {
     } while (response.data['result']['deals'].length > 0);
   }
 
-  /**
-   * @param client
-   * @param lastDeal
-   */
-  /* Temporarily disabled in favor of filscan for more information
-  private async insertDealFromFilfox(client: string, lastDeal: number): Promise<void> {
-    this.logger.debug('Inserting new deals from filfox', { client, lastDeal });
-    let page = 0;
-    let response;
-    do {
-      let breakOuter = false;
-      // Exponential retry as filfox can throttle us
-      response = await retry(
-        async () => {
-          const url = `https://filfox.info/api/v1/deal/list?address=${client}&pageSize=100&page=${page}`;
-          this.logger.debug(`Fetching from ${url}`);
-          let r;
-          try {
-            r = await axios.get(url);
-          } catch (e) {
-            this.logger.warn(e);
-            throw e;
-          }
-          return r;
-        }, {
-        retries: 3,
-        minTimeout: 60_000
-      }
-      );
-      this.logger.debug(`Received ${response.data['deals'].length} deal entries.`);
-      for (const deal of response.data['deals']) {
-        if (deal['id'] <= lastDeal) {
-          breakOuter = true;
-          break;
-        }
-        await Datastore.DealStateModel.updateOne({
-          dealId: deal['id']
-        }, {
-          $setOnInsert: {
-            client,
-            provider: deal['provider'],
-            dealId: deal['id'],
-            state: 'published'
-          }
-        }, {
-          upsert: true
-        });
-      }
-      if (breakOuter) {
-        break;
-      }
-      page += 1;
-    } while (response.data['deals'].length > 0);
-  }
-  */
   private async markExpiredDeals (client: string): Promise<void> {
     const chainHeight = HeightFromCurrentTime() - 120;
     let modified = (await Datastore.DealStateModel.updateMany({
@@ -254,37 +199,37 @@ export default class DealTrackingService extends BaseService {
         await Datastore.DealStateModel.findByIdAndUpdate(dealState.id, {
           state: 'slashed'
         });
-        return;
-      }
-      const result = response.data.result;
-      const expiration: number = result.Proposal.EndEpoch;
-      const slashed = result.State.SlashEpoch > 0;
-      const pieceCid = result.Proposal.PieceCID['/'];
-      const dealActive = result.State.SectorStartEpoch > 0;
-      if (slashed) {
-        await Datastore.DealStateModel.findByIdAndUpdate(dealState.id, {
-          pieceCid, expiration, state: 'slashed'
-        });
-        this.logger.warn(`Deal ${dealState.dealId} is slashed.`);
-      } else if (dealActive) {
-        await Datastore.DealStateModel.findByIdAndUpdate(dealState.id, {
-          pieceCid, expiration, state: 'active'
-        });
-        this.logger.info(`Deal ${dealState.dealId} is active on chain.`);
-        if (dealState.dealCid && dealState.dealCid !== '') {
-          await MetricEmitter.Instance().emit({
-            type: 'deal_active',
-            values: {
-              pieceCid: dealState.pieceCid,
-              pieceSize: dealState.pieceSize,
-              dataCid: dealState.dataCid,
-              provider: dealState.provider,
-              client: dealState.client,
-              verified: dealState.verified,
-              duration: dealState.duration,
-              price: dealState.price
-            }
+      } else {
+        const result = response.data.result;
+        const expiration: number = result.Proposal.EndEpoch;
+        const slashed = result.State.SlashEpoch > 0;
+        const pieceCid = result.Proposal.PieceCID['/'];
+        const dealActive = result.State.SectorStartEpoch > 0;
+        if (slashed) {
+          await Datastore.DealStateModel.findByIdAndUpdate(dealState.id, {
+            pieceCid, expiration, state: 'slashed'
+          });
+          this.logger.warn(`Deal ${dealState.dealId} is slashed.`);
+        } else if (dealActive) {
+          await Datastore.DealStateModel.findByIdAndUpdate(dealState.id, {
+            pieceCid, expiration, state: 'active'
           });
+          this.logger.info(`Deal ${dealState.dealId} is active on chain.`);
+          if (dealState.dealCid && dealState.dealCid !== '') {
+            await MetricEmitter.Instance().emit({
+              type: 'deal_active',
+              values: {
+                pieceCid: dealState.pieceCid,
+                pieceSize: dealState.pieceSize,
+                dataCid: dealState.dataCid,
+                provider: dealState.provider,
+                client: dealState.client,
+                verified: dealState.verified,
+                duration: dealState.duration,
+                price: dealState.price
+              }
+            });
+          }
         }
       }
     }