title: Node.js long process ran twice
body: <p>I have a Node.js restful API built in express.js framework. It is usually hosted by pm2.</p>
  
  <p>One of the services has very long process. When front end called the service, the process started up. Since there is an error in database, the process won't be done properly and the error would be caught. However, before the process reached the error, another exactly same process started with same parameters. So in the meantime, two processes were both running while one was ahead of the other. After a long time, the first process reached error point and returned error. Then the second one returned exactly the same thing.</p>
  
  <p>I checked front end Network and noticed there was actually only one request sent. Where did the second request come from?</p>
  
  <p><strong>Edit 1:</strong></p>
  
  <p>The whole process is: <strong>first process sends query to db -> long time wait -> second process starts up -> second process sends query to db -> long time wait -> first process receives db response -> long time wait -> second process receives db response</strong></p>
  
  <p><strong>Edit 2:</strong></p>
  
  <p>The code of the service is as follow:</p>
  
  <pre><code>import { Express, Request, Response } from "express";
  import * as multer from "multer";
  import * as fs from "fs";
  import { Readable, Duplex } from "stream";
  import * as uid from "uid";
  import { Client } from "pg";
  import * as gdal from "gdal";
  import * as csv from "csv";
  
  import { SuccessPayload, ErrorPayload } from "../helpers/response";
  import { postgresQuery } from "../helpers/database";
  import Config from "../config";
  
  export default class ShapefileRoute {
      constructor(app: Express) {
  
          // Upload a shapefile
          /**
            * @swagger
            * /shapefile:
            *   post:
            *     description: Returns the homepage
            *     responses:
            *       200:
            */
          app.post("/shapefile", (req: Request, res: Response, next: Function): void =&gt; {
  
              // Create instance of multer
              const multerInstance = multer().array("files");
  
              multerInstance(req, res, (err: Error) =&gt; {
                  if (err) {
                      let payload: ErrorPayload = {
                          code: 4004,
                          errorMessage: "Multer upload file error.",
                          errorDetail: err.message,
                          hints: "Check error detail"
                      };
  
                      req.reservePayload = payload;
  
                      next();
  
                      return;
                  }
                  // Extract files
                  let files: any = req.files;
  
                  // Extract body
                  let body: any = JSON.parse(req.body.filesInfo);
  
                  // Other params
                  let writeFilePromises: Promise&lt;any&gt;[] = [];
  
                  let copyFilePromises: Promise&lt;any&gt;[] = [];
  
                  let rootDirectory: string = Config.uploadRoot;
  
                  let outputId: string = uid(4);
  
                  // Reset index of those files
                  let namesIndex: string[] = [];
                  files.forEach((item: Express.Multer.File, index: number) =&gt; {
                      if(item.originalname.split(".")[1] === "csv" || item.originalname.split(".")[1] === "txt" || item.originalname.split(".")[1] === "shp") {
                          namesIndex.push(item.originalname);
                      }
                  })
  
                  // Process and write all files to disk
                  files.forEach((item: Express.Multer.File, outterIndex: number) =&gt; {
                      if(item.originalname.split(".")[1] === "csv" || item.originalname.split(".")[1] === "txt") {
                          namesIndex.forEach((indexItem, index) =&gt; {
                              if(indexItem === item.originalname) {
                                  ShapefileRoute.csv(item, index, writeFilePromises, body, rootDirectory, outputId,);
                              }
                          })
                      } else if (item.originalname.split(".")[1] === "shp") {
                          namesIndex.forEach((indexItem, index) =&gt; {
                              if(indexItem === item.originalname) {
                                  ShapefileRoute.shp(item, index, writeFilePromises, body, rootDirectory, outputId,);
                              }
                          })
                      } else {
                          ShapefileRoute.shp(item, outterIndex, writeFilePromises, body, rootDirectory, outputId,);
                      }
                  })
  
                  // Copy files from disk to database
                  ShapefileRoute.copyFiles(req, res, next, writeFilePromises, copyFilePromises, req.reserveSuperPg, () =&gt; {
                      ShapefileRoute.loadFiles(req, res, next, copyFilePromises, body, outputId)
                  });
  
  
              })
          });
      }
  
      // Process csv file
      static csv(file: Express.Multer.File, index: number, writeFilePromises: Promise&lt;any&gt;[], body: any, rootDirectory: string, outputId: string) {
  
          // Streaming file to pivotcsv
          writeFilePromises.push(new Promise((resolve, reject) =&gt; {
  
              // Get specification from body
              let delimiter: string;
              let spec: any;
              let lrsColumns: string[] = [null, null, null, null, null, null];
              body.layers.forEach((jsonItem, i) =&gt; {
                  if (jsonItem.name === file.originalname.split(".")[0]) {
                      delimiter = jsonItem.file_spec.delimiter;
                      spec = jsonItem
                      jsonItem.lrs_cols.forEach((lrsCol) =&gt; {
                          switch(lrsCol.lrs_type){
                              case "rec_id":
                              lrsColumns[0] = lrsCol.name;
                              break;
                              case "route_id":
                              lrsColumns[1] = lrsCol.name;
                              break;
                              case "f_meas":
                              lrsColumns[2] = lrsCol.name;
                              break;
                              case "t_meas":
                              lrsColumns[3] = lrsCol.name;
                              break;
                              case "b_date":
                              lrsColumns[4] = lrsCol.name;
                              break;
                              case "e_date":
                              lrsColumns[5] = lrsCol.name;
                              break;
                          }
                      })
                  }
              });
  
              // Pivot csv file
              ShapefileRoute.pivotCsv(file.buffer, `${rootDirectory}/${outputId}_${index}`, index, delimiter, outputId, lrsColumns, (path) =&gt; {
                  console.log("got pivotCsv result");
                  spec.order = index;
                  resolve({
                      path: path,
                      spec: spec
                  });
              }, reject);
          }));
      }
  
      // Process shapefile
      static shp(file: Express.Multer.File, index: number, writeFilePromises: Promise&lt;any&gt;[], body: any, rootDirectory: string, outputId: string) {
  
          // Write file to disk and then call shp2csv to gennerate csv
          writeFilePromises.push(new Promise((resolve, reject) =&gt; {
  
              // Write shpefile to disk
              fs.writeFile(`${rootDirectory}/shps/${file.originalname}`, file.buffer, (err) =&gt; {
  
                  // If it is .shp file, resolve it's path and spec
                  if(file.originalname.split(".")[1] === "shp") {
                      // Find spec of the shapefile from body
                      body.layers.forEach((jsonItem, i) =&gt; {
                          if (jsonItem.name === file.originalname.split(".")[0]) {
                              let recordColumn: string = null;
                              let routeIdColumn: string = null;
                              jsonItem.lrs_cols.forEach((lrsLayer) =&gt; {
                                  if (lrsLayer.lrs_type === "rec_id") {
                                      recordColumn = lrsLayer.name;
                                  }
                                  if (lrsLayer.lrs_type === "route_id") {
                                      routeIdColumn = lrsLayer.name;
                                  }
                              })
  
                              // Transfer shp to csv
                              ShapefileRoute.shp2csv(`${rootDirectory}/shps/${file.originalname}`, `${rootDirectory}/${outputId}_${index}`, index, outputId, recordColumn, routeIdColumn, (path, srs) =&gt; {
  
                                  // Add coordinate system, geom column and index of this file to spec
                                  jsonItem.file_spec.proj4 = srs;
                                  jsonItem.file_spec.geom_col = "geom";
                                  jsonItem.order = index;
  
                                  // Return path and spec
                                  resolve({
                                      path: path,
                                      spec: jsonItem
                                  })
                              }, (err) =&gt; {
                                  reject;
                              })
                          }
                      });
                  } else {
                      resolve(null);
                  }
              })
          }));
      }
  
      // Copy files to database
      static copyFiles(req: Request, res: Response, next: Function, writeFilePromises: Promise&lt;any&gt;[], copyFilePromises: Promise&lt;any&gt;[], client: Client, callback: () =&gt; void) {
          // Take all files generated by writefile processes
          Promise.all(writeFilePromises)
          .then((results) =&gt; {
  
              // Remove null results. They are from .dbf .shx etc of shapefile.
              const files: any = results.filter(arr =&gt; arr);
  
              // Create promise array. This will be triggered after all files are written to database.
              files.forEach((file) =&gt; {
                  copyFilePromises.push(new Promise((copyResolve, copyReject) =&gt; {
                      let query: string = `copy lbo.lbo_temp from '${file.path}' WITH NULL AS 'null';`;
  
                      // Create super user call
                      postgresQuery(client, query, (data) =&gt; {
                          copyResolve(file.spec);
                      }, copyReject);
                  }));
              });
  
              // Trigger upload query
              callback()
          })
          .catch((err) =&gt; {
              // Response as error if any file generating is wrong
              let payload: ErrorPayload = {
                  code: 4004,
                  errorMessage: "Something wrong when processing csv and/or shapefile.",
                  errorDetail: err.message,
                  hints: "Check error detail"
              };
  
              req.reservePayload = payload;
  
              next();
  
          })
      }
  
  
      // Load layers in database
      static loadFiles(req: Request, res: Response, next: Function, copyFilePromises: Promise&lt;any&gt;[], body: any, outputId: string) {
  
          Promise.all(copyFilePromises)
          .then((results) =&gt; {
  
              // Resort all results by the order assigned when creating files
              results.sort((a, b) =&gt; {
                  return a.order - b.order;
              });
              results.forEach((result) =&gt; {
                  delete result.order;
              });
  
              // Create JSON for load layer database request
              let taskJson = body;
              taskJson.layers = results;
              let query: string = `select lbo.load_layers2(p_session_id := '${outputId}', p_layers := '${JSON.stringify(taskJson)}'::json)`;
  
              postgresQuery(req.reservePg, query, (data) =&gt; {
                  // Get result
                  let result = data.rows[0].load_layers2.result;
  
                  // Return 4003 error if no result
                  if (!result) {
                      let payload: ErrorPayload = {
                          code: 4003,
                          errorMessage: "Load layers error.",
                          errorDetail: data.rows[0].load_layers2.error ? data.rows[0].load_layers2.error.message : "Load layers returns no result.",
                          hints: "Check error detail"
                      };
  
                      req.reservePayload = payload;
  
                      next();
  
                      return;
                  }
                  let payload: SuccessPayload = {
                      type: "string",
                      content: "Upload files done."
                  };
  
                  req.reservePayload = payload;
  
                  next();
              }, (err) =&gt; {
                  req.reservePayload = err;
  
                  next();
              });
  
          })
          .catch((err) =&gt; {
              // Response as error if any file generating is wrong
              let payload: ErrorPayload = {
                  code: 4004,
                  errorMessage: "Something wrong when copy files to database.",
                  errorDetail: err,
                  hints: "Check error detail"
              };
  
              req.reservePayload = payload;
  
              next();
          })
  
      }
  
      // Pivot csv process. Write output csv to disk and return path of the file.
      static pivotCsv(buffer: Buffer, outputPath: string, inputIndex: number, delimiter: string, outputId: string, lrsColumns: string[], callback: (path: string) =&gt; void, errCallback: (err: Error) =&gt; void) {
  
          let inputStream: Duplex = new Duplex();
  
          // Define output stream
          let output = fs.createWriteStream(outputPath, {flags: "a"});
          // Callback when output stream is done
          output.on("finish", () =&gt; {
              console.log("output stream finish");
              callback(outputPath);
          });
  
          // Define parser stream
          let parser = csv.parse({
              delimiter: delimiter
          });
          // Close output stream when parser stream is end
          parser.on("end", () =&gt; {
              console.log("parser stream end");
              output.end();
          });
          // Write data when a chunck is parsed
          let header = [null, null, null, null, null, null];
          let attributesHeader = [];
          let i = 0;
          let datumIndex: boolean = true;
          parser.on("data", (chunk) =&gt; {
              console.log("parser received on chunck: ", i);
              if (datumIndex) {
                  chunk.forEach((datum, index) =&gt; {
                      if (lrsColumns.includes(datum)) {
                          header[lrsColumns.indexOf(datum)] = index;
                      } else {
                          attributesHeader.push({
                              name: datum,
                              index: index
                          })
                      }
                  });
                  datumIndex = false;
              } else {
                  i ++;
                  // let layer_id = ;
                  let rec_id = header[0] ? chunk[header[0]] : i;
                  let route_id = header[1] ? chunk[header[1]] : null;
                  let f_meas = header[2] ? chunk[header[2]] : null;
                  let t_meas = header[3] ? chunk[header[3]] : null;
                  let b_date = header[4] ? chunk[header[4]] : null;
                  let e_date = header[5] ? chunk[header[5]] : null;
  
                  let attributes = {};
  
                  attributesHeader.forEach((attribute) =&gt; {
                      attributes[attribute.name] = chunk[attribute.index];
                  });
  
                  let attributesOrdered = {};
                  Object.keys(attributes).sort().forEach((key) =&gt; {
                      attributesOrdered[key] = attributes[key];
                  });
  
                  let outputData = `${outputId}\t${inputIndex}\t${rec_id}\t${route_id}\tnull\t${f_meas}\t${t_meas}\t${b_date}\t${e_date}\tnull\t${JSON.stringify(attributesOrdered)}\n`;
  
                  output.write(outputData);
              }
          });
  
  
  
          inputStream.push(buffer);
          inputStream.push(null);
          inputStream.pipe(parser);
      }
  
      // Write shp and transfer to database format. Return file path and projection.
      static shp2csv(inputPath: string, outputPath: string, i: number, ouputId: string, recordColumn: string, routeIdColumn: string, callback: (path: string, prj: string) =&gt; void, errCallback: (err: Error) =&gt; void) {
          let dataset = gdal.open(inputPath);
          let layercount = dataset.layers.count();
          let layer = dataset.layers.get(0);
          let output = fs.createWriteStream(outputPath, {flags: "a"});
          output.on("finish", () =&gt; {
              callback(outputPath, layer.srs.toProj4());
          });
          layer.features.forEach((feature, featureId) =&gt; {
              let geom;
              let recordId: number = null;
              let routeId: string = null;
              try {
                  let geomWKB = feature.getGeometry().toWKB();
                  let geomWKBString = geomWKB.toString("hex");
                  geom = geomWKBString;
                  if (recordColumn) {
                      recordId = feature.fields.get(recordColumn);
                  }
                  if (routeIdColumn) {
                      routeId = feature.fields.get(routeIdColumn);
                  }
              }
              catch (err) {
                  console.log(err);
              }
              let attributes = {};
  
              let attributesOrdered = {};
  
              feature.fields.forEach((value, field) =&gt; {
                  if (field != recordColumn &amp;&amp; field != routeIdColumn) {
                      attributes[field] = value;
                  }
              });
  
              Object.keys(attributes).sort().forEach((key) =&gt; {
                  attributesOrdered[key] = attributes[key];
              });
              output.write(`${ouputId}\t${i.toString()}\t${recordId ? recordId : (featureId + 1).toString()}\t${routeId}\tnull\tnull\tnull\tnull\tnull\t${geom}\t${JSON.stringify(attributesOrdered)}\n`);
          });
          output.end();
      }
  }
  </code></pre>