cartodb/CartoDB-SQL-API
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #680 from CartoDB/feature/ch139802/phl-inconsistent-errors-when-downloading

Browse Source 
[ch139802] Avoid concurrent requests from getting stuck when exporting file formats handled by OGR
This commit is contained in:
Shylpx 2021-03-29 10:57:29 +01:00 committed by GitHub
commit 63e3f64fdf
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 62 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

150
lib/models/formats/ogr.js
View File

@ -1,15 +1,12 @@
'use strict';
var crypto = require('crypto');
var uuid = require('uuid');
var step = require('step');
var fs = require('fs');
var _ = require('underscore');
var PSQL = require('cartodb-psql');
var spawn = require('child_process').spawn;
// Keeps track of what's waiting baking for export
var bakingExports = {};
function OgrFormat (id) {
this.id = id;
}
@ -32,15 +29,6 @@ OgrFormat.prototype = {
getFileExtension: function () { return this._fileExtension; },
getKey: function (options) {
return [this.id,
options.dbopts.dbname,
options.dbopts.user,
options.gn,
this.generateMD5(options.filename),
this.generateMD5(options.sql)].concat(options.skipfields).join(':');
},
generateMD5: function (data) {
var hash = crypto.createHash('md5');
hash.update(data);
@ -241,7 +229,8 @@ OgrFormat.prototype.toOGR_SingleFile = function (options, fmt, callback) {
userId,
gcol,
this.generateMD5(layername),
this.generateMD5(sql)
this.generateMD5(sql),
uuid.v4()
].concat(skipfields).join(':'));
var outdirpath = tmpdir + '/sqlapi-' + process.pid + '-' + reqKey;
var dumpfile = outdirpath + ':cartodb-query.' + ext;
@ -252,51 +241,31 @@ OgrFormat.prototype.toOGR_SingleFile = function (options, fmt, callback) {
};
OgrFormat.prototype.sendResponse = function (opts, callback) {
// var next = callback;
var reqKey = this.getKey(opts);
var qElem = new ExportRequest(opts.sink, callback, opts.beforeSink);
var baking = bakingExports[reqKey];
if (baking) {
baking.req.push(qElem);
} else {
baking = bakingExports[reqKey] = { req: [qElem] };
this.generate(opts, function (err, dumpfile) {
if (opts.profiler) {
opts.profiler.done('generate');
}
step(
function sendResults () {
var nextPipe = function (finish) {
var r = baking.req.shift();
if (!r) { finish(null); return; }
r.sendFile(err, dumpfile, function () {
nextPipe(finish);
});
};
var exportRequest = new ExportRequest(opts.sink, callback, opts.beforeSink);
if (!err) {
nextPipe(this);
} else {
_.each(baking.req, function (r) {
r.cb(err);
});
return true;
}
},
function cleanup (/* err */) {
delete bakingExports[reqKey];
// unlink dump file (sync to avoid race condition)
console.log('removing', dumpfile);
try { fs.unlinkSync(dumpfile); } catch (e) {
if (e.code !== 'ENOENT') {
console.log('Could not unlink dumpfile ' + dumpfile + ': ' + e);
}
this.generate(opts, function (err, dumpfile) {
if (opts.profiler) {
opts.profiler.done('generate');
}
step(
function sendResult () {
if (err) {
exportRequest.cb(err);
return true;
}
exportRequest.sendFile(dumpfile, this);
},
function cleanup () {
// unlink dump file (sync to avoid race condition)
console.log('removing', dumpfile);
try { fs.unlinkSync(dumpfile); } catch (e) {
if (e.code !== 'ENOENT') {
console.log('Could not unlink dumpfile ' + dumpfile + ': ' + e);
}
}
);
});
}
}
);
});
};
function ExportRequest (ostream, callback, beforeSink) {
@ -305,6 +274,7 @@ function ExportRequest (ostream, callback, beforeSink) {
this.ostream = ostream;
this.istream = null;
this.canceled = false;
this.read = false;
var that = this;
@ -316,42 +286,46 @@ function ExportRequest (ostream, callback, beforeSink) {
});
}
ExportRequest.prototype.sendFile = function (err, filename, callback) {
if (err) {
return callback(err);
ExportRequest.prototype.sendFile = function (filename, callback) {
if (this.canceled) {
return callback();
}
var that = this;
if (!this.canceled) {
this.istream = fs.createReadStream(filename)
.on('open', function (/* fd */) {
if (that.beforeSink) {
that.beforeSink();
}
that.istream
.pipe(that.ostream)
.on('end', () => {
callback();
that.cb();
})
.on('error', (err) => {
callback();
that.cb(err);
});
})
.on('error', function (e) {
console.log("Can't send response: " + e);
that.ostream.end();
that.cb(e);
this.istream = fs.createReadStream(filename)
.on('open', function () {
if (that.beforeSink) {
that.beforeSink();
}
that.istream
.pipe(that.ostream)
.on('end', () => {
that.cb();
callback();
})
.on('error', (err) => {
that.cb(err);
callback();
});
})
.on('error', function (e) {
console.log("Can't send response: " + e);
that.read = true;
that.ostream.end();
that.cb(e);
callback();
})
.on('end', () => {
that.read = true;
that.cb();
callback();
})
.on('close', () => {
if (!that.read) { // NOTE: Covering 304 responses, when the 'ostream' is closed after 'pipe'
callback();
})
.on('end', () => {
that.cb();
callback();
});
} else {
callback();
}
}
});
};
module.exports = OgrFormat;