FIX scheduling mismatch with mileseconds

Author: manobi

examples/failed.js

@@ -8,6 +8,7 @@ manobi.addJob('compress', async (ctx, done) => {
 
 
 (async () => {
+    await manobi.start();
     return createTask({
         date: new Date().toISOString(),
         job: 'compress',

examples/fast.js

@@ -1,20 +1,30 @@
-const fetch = require('node-fetch'),
-    { worker, createTask } = require('..')(),
+const { worker, createTask } = require('..')(),
     manobi = worker('manobi-fast');
 
 manobi.addJob('compress', (ctx, done) => {
-    done(null, {sucesso: 'muleque'});
+    done(null, {noerrorbaby: true});
 });
 
 (async () => {
     await manobi.start()
-    await createTask({
+    /*await createTask({
         date: new Date().toISOString(),
         job: 'compress',
+        status: 'waiting',
         worker: 'manobi-fast',
         repeat: {
-            interval: {seconds: 1}
+            limit: 10,
+            interval: {
+                seconds: 6
+            }
+        },
+        repeat: {
+            limit: 10,
+            interval: {
+                seconds: 1
+            }
         }
     });
+    */
     //await manobi.stop()
 })();

lib/context.js

@@ -36,7 +36,6 @@ module.exports = class Context {
     }
 
     startTimeout(){
-        const ms = this.task.timeout || 5000;
         this.timeout = setTimeout(() => {
             this.cancel();
             this.done({error: 'Job timed out'});
@@ -47,6 +46,7 @@ module.exports = class Context {
      * @param {Function} job - The job function to be executed
      */
     start(job){
+        console.log(new Date().toISOString(), 'Started Running', this.task.id, 'date', this.task.date);
         performance.mark('job_start');
         // Defines a timeout for the job
         this.startTimeout();
@@ -89,6 +89,7 @@ module.exports = class Context {
      * @returns {Number} duration - Time elapsed from start to stop in ms
      */
     stop(){
+        console.log(new Date().toISOString(), 'Stoping');
         this.running = undefined;
         clearTimeout(this.timeout);
         performance.mark('job_stop');
@@ -102,13 +103,12 @@ module.exports = class Context {
      * @param {Number} current - Current state of property like repeat.limit=100
      * @returns {Boolean} - rerunable or not
      */
-    rerunable(attr, current){
+    rerunable(attr, current = 0){
         const {task} = this,
             node = task[attr] || {},
             interval = node.interval,
             limit = node.limit || Infinity,
             canRerun = current < limit;
-        
         if(node && interval && canRerun){
             return delay(task.date, interval);
         }
@@ -135,12 +135,16 @@ module.exports = class Context {
      * @private
      * @param {Object | null} err 
      * @param {Object | null} res 
-     * @returns {String} - done, complete, retry, fail
+     * @returns {String} - done, complete, retry, failed
      */
     nextStatus(err, res){
-        const success = this.repeatable() ? 'done' : 'complete';
-        const fail = this.retryable() ? 'retry' : 'failed';
-        return err ? fail : success;
+        const success = () => {
+            return this.repeatable() ? 'done' : 'complete';
+        }
+        const fail = () => {
+            return this.retryable() ? 'retry' : 'failed';
+        }
+        return err ? fail() : success();
     }
 
     /**

lib/couchdb-adaptor.js

@@ -79,7 +79,6 @@ module.exports = class Adaptor {
         this.feed = this.db.follow({
             include_docs: true,
             since: 'now',
-            heartbeat: 1000,
             /*
             * TODO: use mango query
             */
@@ -108,7 +107,6 @@ module.exports = class Adaptor {
     * @method unsubscribe
     */
     unsubscribe(){
-        this.feed.off('change', this._change);
         return this.feed.stop();
     }
 
@@ -147,6 +145,8 @@ module.exports = class Adaptor {
             _id,
         };
         delete document.id;
+        console.log(new Date().toISOString(), 'sending update')
+        console.log('----------------------------------');
         return this.db.insert(document);
     }
 

lib/scheduler.js

@@ -13,21 +13,22 @@ module.exports = class Scheduler {
     }
 
     schedule(task, callback){
-        const date = Date.parse(task.date);
+        const date = new Date(Date.parse(task.date)).setMilliseconds(0)
+        // Remove previous scheduled task
+        this.unschedule(task.id);
 
         if(Scheduler.isOutdated(date)){
+            console.log(new Date().toISOString(), 'executing outdated', task.id, 'date', task.date);
             return callback();
         }
 
-        // Remove previous scheduled task
-        this.unschedule(task.id);
-        
         // Assigns a function that calls the job with data as param
         this.scheduled[task.id] = {
             run: callback,
             date: task.date // For logging reasons schedule the original form
         }
-        console.log('Scheduling', task.id, 'date', task.date);
+
+        console.log(new Date().toISOString(), 'Scheduling', task.id, 'date', task.date);
         return this.tasks.once(date, this.scheduled[task.id]['run']);
     }
     /*

lib/worker.js

@@ -21,7 +21,9 @@ module.exports = class Worker extends Scheduler {
         // Setup storage connection
         this.storage = new Adaptor();
         this.storage.name = this.name;
-        this.storage.onChange = (change) => this.changeHandler.call(this, change);
+        this.storage.onChange = (change) => {
+            this.changeHandler.call(this, change);
+        }
         this.storage.connect(options);
     }
 
@@ -91,16 +93,20 @@ module.exports = class Worker extends Scheduler {
          * @memberof Context
          */
         ctx.done = (err, res) => {
-            if(this.finished){
+            console.log(new Date().toISOString(), 'Done called with', err, res);
+            
+            if(ctx.finished){
                 throw new Error('Done called twice');
             }
 
-            this.finished = true;
+            ctx.finished = true;
             ctx.stop();
 
             // Enqueue the task update
             this.enqueue(task.id, _ => {
-                return this.storage.update(ctx.next(err, res));
+                const next = ctx.next(err, res);
+                console.log(new Date().toISOString(), 'Queued updated', task.id, 'date', task.date);
+                return this.storage.update(next);
             });
         }
         return ctx;
@@ -117,7 +123,6 @@ module.exports = class Worker extends Scheduler {
 
         return () => {
             context.start(job);
-            console.log('Running', task.id, 'date', task.date);
         }
     }
 
@@ -129,16 +134,17 @@ module.exports = class Worker extends Scheduler {
         if(!change){
             throw new Error(`Undefined task, it expects an object as argument`);
         }
+        
+        console.log(new Date().toISOString(), 'update received', change.id, 'date', change.date);
 
         ['id'].forEach(attr => {
             if(!change[attr]){
                 throw new Error(`Invalid task, "${attr}" is missing`);
             }
         });
-
-        console.log('Change', change.id, 'date', change.date);
 
         if(change.status === 'cancel'){
+            console.log(new Date().toISOString(), 'task have being deleted', change.id, 'date', change.date);
             return this.unschedule(change.id);
         };
 

spec/scheduler.spec.js

@@ -20,7 +20,7 @@ describe('Scheduler', () => {
         it('Append the task to memory book', () => {
             const now = DateTime.local(),
                 date = now.plus({'seconds': 20}).toISO(),
-                timestamp = DateTime.fromISO(date).toMillis(),
+                timestamp = DateTime.fromISO(date).startOf('second').toMillis(),
                 task = {
                     date,
                     id: '857957d1f1631ac8714b5d1cfd000d39'