Clean old workers and tasks

initializer.js

module.exports = {
    start: function(_api, next) {
        var maxWorkerAge = 10000;
        api.log('Removing stuck workers older than ' + maxWorkerAge + 'ms', 'info');
        api.resque.queue.cleanOldWorkers(maxWorkerAge, function(err, result) {
            if (err) {
                api.log(err, 'error');
            }
            if (Object.keys(result).length > 0) {
                api.log('Removed stuck workers with errors: ', 'info', result);
            }
        });
      
        // Every 60 seconds, remove failed tasks and re-enqueue stuck/hung jobs.
        if (api.config.tasks.scheduler) {
            setInterval(function() {
                api.tasks.failed(0, -1, (failedTasksErr, failedTasks) => {
                    failedTasks.map(task => {
                        api.tasks.removeFailed(task, (removeError, countRemoved) => {
                            if (removeError) {
                                api.log('Error removing failed task', 'error', removeError, task);
                            } else {
                                api.log('Removed failed task', 'info', task);
                            }
                        });
                    });
                });

                // Find recurring jobs that have been hanging around too long without actually executing
                Object.keys(api.tasks.tasks).map(taskName => {
                    let task = api.tasks.tasks[taskName];

                    if (task.frequency > 0) {
                        api.tasks.scheduledAt(task.queue, taskName, {}, (scheduledAtError, timestamps) => {
                            (timestamps || []).map(entry => {
                                let timestamp = +entry,
                                    humanReadable = (new Date(timestamp * 1000));

                                if (timestamp > 0 && timestamp < (Date.now() / 1000) - 300) {
                                    api.log('Found stuck task', 'warning', timestamp, task);
                                    api.tasks.delDelayed(task.queue, taskName, {}, delDelayedError => {
                                        if (delDelayedError) {
                                            api.log('Error clearing stuck task ' + taskName, 'error', delDelayedError);
                                        } else {
                                            api.log('Cleared stuck task ' + taskName + ', stuck since ' + humanReadable, 'info');
                                            api.tasks.enqueueRecurrentJob(taskName, function() {
                                                // Nothing to do here.
                                            });
                                        }
                                    });
                                }
                            });
                        });
                    }
                });
            }, 60000).unref();
        }
    }
};