retropilot
/
retropilot-server
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

fix(worker): fix segment processing/file upload status

pull/4/head
Cameron Clough 2022-03-24 22:52:21 +00:00
parent 07551a7514
commit c59f2a07ea
No known key found for this signature in database
GPG Key ID: BFB3B74B026ED43F
1 changed files with 35 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

71
src/worker/worker.js
View File

@ -210,7 +210,7 @@ async function processSegmentsRecursive() {
logger.error(`FAILING TO PROCESS SEGMENT,${dongleId} ${driveIdentifier} ${segmentId} JSON: ${JSON.stringify(segment)} SKIPPING `); logger.error(`FAILING TO PROCESS SEGMENT,${dongleId} ${driveIdentifier} ${segmentId} JSON: ${JSON.stringify(segment)} SKIPPING `);
segmentProcessPosition += 1; segmentProcessPosition += 1;
} else { } else {
Promise.all([ await Promise.all([
processSegmentRLog(fileStatus['rlog.bz2']), processSegmentRLog(fileStatus['rlog.bz2']),
processSegmentVideo(fileStatus['qcamera.ts']), processSegmentVideo(fileStatus['qcamera.ts']),
]) ])
@ -269,24 +269,24 @@ async function updateSegments() {
}); });
logger.info('updateSegments - total segments', segments.length); logger.info('updateSegments - total segments', segments.length);
if (segments.length > 0) { await Promise.all(segments.map(async (segment) => {
// we process at most 15 segments per batch // we process at most 15 segments per batch
segments.length = Math.min(segments.length, 15); if (segmentProcessQueue.length >= 15) {
logger.debug(`updateSegments - processing ${segments.length} segments`); return;
}
await Promise.all(segments.map(async (segment) => { const {
logger.debug('updateSegments - segment', segment); id,
const { created,
id,
created,
dongle_id: dongleId, dongle_id: dongleId,
drive_identifier: driveIdentifier, drive_identifier: driveIdentifier,
is_processed: isProcessed, is_processed: isProcessed,
segment_id: segmentId, segment_id: segmentId,
} = segment; } = segment;
logger.debug('updateSegments - segment', driveIdentifier, segmentId);
const dongleIdHash = crypto.createHmac('sha256', process.env.APP_SALT) const dongleIdHash = crypto.createHmac('sha256', process.env.APP_SALT)
.update(dongleId) .update(dongleId)
.digest('hex'); .digest('hex');
const driveIdentifierHash = crypto.createHmac('sha256', process.env.APP_SALT) const driveIdentifierHash = crypto.createHmac('sha256', process.env.APP_SALT)
.update(driveIdentifier) .update(driveIdentifier)
@ -306,27 +306,27 @@ async function updateSegments() {
dcamera: 'dcamera.hevc', dcamera: 'dcamera.hevc',
qcamera: 'qcamera.ts', qcamera: 'qcamera.ts',
qlog: 'qlog.bz2', qlog: 'qlog.bz2',
rlog: 'rlog.bz2', rlog: 'rlog.bz2',
}; };
const fileStatus = { const fileStatus = {
[SegmentFiles.fcamera]: false, [SegmentFiles.fcamera]: undefined,
[SegmentFiles.dcamera]: false, [SegmentFiles.dcamera]: undefined,
[SegmentFiles.qcamera]: false, [SegmentFiles.qcamera]: undefined,
[SegmentFiles.qlog]: false, [SegmentFiles.qlog]: undefined,
[SegmentFiles.rlog]: false, [SegmentFiles.rlog]: undefined,
}; };
directoryTree.children.forEach((file) => { directoryTree.children.forEach((file) => {
if (file.name in fileStatus) { if (file.name in fileStatus) {
logger.debug('updateSegments - found file', file.name); logger.debug('updateSegments - found file', file.name);
fileStatus[file.name] = true; fileStatus[file.name] = file.path;
} }
}); });
const uploadComplete = Object.keys(fileStatus).every((key) => fileStatus[key]); const uploadComplete = Object.keys(fileStatus).every((key) => !!fileStatus[key]);
logger.debug('updateSegments - uploadComplete', uploadComplete); logger.debug('updateSegments - uploadComplete', uploadComplete);
if (fileStatus[SegmentFiles.qcamera] && fileStatus[SegmentFiles.rlog] && !isProcessed) { if (fileStatus[SegmentFiles.qcamera] && fileStatus[SegmentFiles.rlog] && !isProcessed) {
// can process // can process
logger.debug('updateSegments - can process', id); logger.debug('updateSegments - can process', id);
segmentProcessQueue.push({ segmentProcessQueue.push({
@ -350,10 +350,9 @@ async function updateSegments() {
await DriveSegments.update({ await DriveSegments.update({
is_stalled: true, is_stalled: true,
}, { where: { id } }); }, { where: { id } });
} }
})); }));
}
if (segmentProcessQueue.length > 0) { if (segmentProcessQueue.length > 0) {
logger.info('updateSegments - processing', segmentProcessQueue.length); logger.info('updateSegments - processing', segmentProcessQueue.length);