Changes to how tasks are handled to allow better fault handling

acs.coffee

@@ -93,27 +93,39 @@ updateDevice = (currentRequest, actions, callback) ->
 
 runTask = (currentRequest, task, methodResponse) ->
   cwmpResponse = {}
-  actions = tasks.task(task, methodResponse, cwmpResponse)
+  deviceUpdates = {}
+  status = tasks.task(task, methodResponse, cwmpResponse, deviceUpdates)
 
-  updateDevice(currentRequest, actions)
+  updateDevice(currentRequest, deviceUpdates)
 
-  if Object.keys(cwmpResponse).length > 0
+  switch status
+    when tasks.STATUS_FINISHED
+      db.memcached.del(String(task._id))
+      db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, {safe: true}, (err, removed) ->
+        util.log("#{currentRequest.deviceId}: Completed task #{task.name}(#{task._id})")
+        nextTask(currentRequest)
+      )
+    when tasks.STATUS_FAULT
+      db.saveTask(task, (err) ->
+        # Faulty task. No more work to do until task is deleted.
+        res = tr069.response(null, cwmpResponse)
+        writeResponse(currentRequest, res)
+      )
+    when tasks.STATUS_PENDING
+      db.saveTask(task, (err) ->
+        # task expects CPE confirmation later
+        nextTask(currentRequest)
+      )
+    when tasks.STATUS_STARTED
       db.updateTask(task, (err) ->
         res = tr069.response(task._id, cwmpResponse)
         writeResponse(currentRequest, res)
       )
-    return
+    else
-
+      throw Error('Unknown task status')
-  # Task finished
-  db.memcached.del(String(task._id))
-  db.tasksCollection.remove({'_id' : mongodb.ObjectID(String(task._id))}, {safe: true}, (err, removed) ->
-    util.log("#{currentRequest.deviceId}: Completed task #{task.name}(#{task._id})")
-    cookies = {}
-    nextTask(currentRequest, cookies)
-  )
 
 
-nextTask = (currentRequest, cookies) ->
+nextTask = (currentRequest) ->
   cur = db.tasksCollection.find({'device' : currentRequest.deviceId}).sort(['timestamp']).limit(1)
   cur.nextObject( (err, task) ->
     cwmpResponse = {}
@@ -136,18 +148,17 @@ nextTask = (currentRequest, cookies) ->
               )
             else
               # no discrepancy, return empty response
-              res = tr069.response(null, cwmpResponse, cookies)
+              res = tr069.response(null, cwmpResponse)
               writeResponse(currentRequest, res)
           )
         else
           # no discrepancy, return empty response
-          res = tr069.response(null, cwmpResponse, cookies)
+          res = tr069.response(null, cwmpResponse)
           writeResponse(currentRequest, res)
       )
-      return
     else if task.fault?
       # last task was faulty. Do nothing until until task is deleted
-      res = tr069.response(null, cwmpResponse, cookies)
+      res = tr069.response(null, cwmpResponse)
       writeResponse(currentRequest, res)
     else
       util.log("#{currentRequest.deviceId}: Started task #{task.name}(#{task._id})")
@@ -244,21 +255,23 @@ else
 
         db.getTask(taskId, (task) ->
           if not task
-            nextTask(currentRequest, cookies)
+            nextTask(currentRequest)
           else
             runTask(currentRequest, task, cwmpRequest.methodResponse)
         )
       else if cwmpRequest.fault?
         taskId = cwmpRequest.id
         util.log("#{currentRequest.deviceId}: Fault response for task #{taskId}")
-        db.tasksCollection.update({_id : mongodb.ObjectID(taskId)}, {$set : {fault : cwmpRequest.fault}}, (err) ->
+
-          # Faulty task. No more work to do until task is deleted.
+        db.getTask(taskId, (task) ->
-          res = tr069.response(null, cwmpResponse, cookies)
+          if not task
-          writeResponse(currentRequest, res)
+            nextTask(currentRequest)
+          else
+            runTask(currentRequest, task, cwmpRequest.fault)
         )
       else
         # cpe sent empty response. start sending acs requests
-        nextTask(currentRequest, cookies)
+        nextTask(currentRequest)
 
 
   server.listen config.ACS_PORT, config.ACS_INTERFACE

db.coffee

@@ -56,6 +56,14 @@ updateTask = (task, callback) ->
   )
 
 
+saveTask = (task, callback) ->
+  task._id = mongodb.ObjectID(String(task._id))
+  tasksCollection.save(task, (err) ->
+    callback(err)
+  )
+
+
 exports.memcached = memcached
 exports.getTask = getTask
 exports.updateTask = updateTask
+exports.saveTask = saveTask

tasks.coffee

@@ -1,37 +1,64 @@
 common = require './common'
 BATCH_SIZE = 16
 
-this.init = (task, methodResponse, cwmpResponse) ->
+exports.STATUS_QUEUED = STATUS_QUEUED = 0
-  ret = {}
+exports.STATUS_STARTED = STATUS_STARTED = 1
+exports.STATUS_PENDING = STATUS_PENDING = 2
+exports.STATUS_FAULT = STATUS_FAULT = 3
+exports.STATUS_FINISHED = STATUS_FINISHED = 4
+
+
+this.init = (task, methodResponse, cwmpResponse, deviceUpdates) ->
+  if methodResponse.faultcode?
+    task.fault = methodResponse
+    return STATUS_FAULT
+
   if not task.subtask?
     task.subtask = {'name' : 'getParameterNames', 'parameterPath' : '', 'nextLevel' : false}
 
   if task.subtask.name == 'getParameterNames'
-    common.extend(ret, this.getParameterNames(task.subtask, methodResponse, cwmpResponse))
+    status = this.getParameterNames(task.subtask, methodResponse, cwmpResponse, deviceUpdates)
-    if ret.parameterNames? # task finished
+    if status is STATUS_FINISHED
       parameterNames = []
-      for p in ret.parameterNames
+      for p in deviceUpdates.parameterNames
         if not common.endsWith(p[0], '.')
           parameterNames.push(p[0])
       task.subtask = {'name' : 'getParameterValues', 'parameterNames' : parameterNames}
+    else if status != STATUS_STARTED
+      throw Error('Unexpected subtask status')
 
   if task.subtask.name == 'getParameterValues'
-    common.extend(ret, this.getParameterValues(task.subtask, methodResponse, cwmpResponse))
+    status = this.getParameterValues(task.subtask, methodResponse, cwmpResponse, deviceUpdates)
+    if status is STATUS_FINISHED
+      return STATUS_FINISHED
+    else if status is not STATUS_STARTED
+      throw Error('Unexpected subtask status')
 
-  ret
+  return STATUS_STARTED
+
+
+this.getParameterNames = (task, methodResponse, cwmpResponse, deviceUpdates) ->
+  if methodResponse.faultcode?
+    task.fault = methodResponse
+    return STATUS_FAULT
 
-this.getParameterNames = (task, methodResponse, cwmpResponse) ->
   if methodResponse.type is 'GetParameterNamesResponse'
-    return {'parameterNames' : methodResponse.parameterList}
+    deviceUpdates.parameterNames = methodResponse.parameterList
-  else
+    return STATUS_FINISHED
+
   cwmpResponse.methodRequest = {
     type : 'GetParameterNames',
     parameterPath : task.parameterPath,
     nextLevel : if task.nextLevel? then task.nextLevel else false
   }
-  return
+  return STATUS_STARTED
+
+
+this.getParameterValues = (task, methodResponse, cwmpResponse, deviceUpdates) ->
+  if methodResponse.faultcode?
+    task.fault = methodResponse
+    return STATUS_FAULT
 
-this.getParameterValues = (task, methodResponse, cwmpResponse) ->
   if not task.currentIndex?
     task.currentIndex = 0
   else if methodResponse.parameterList?
@@ -40,17 +67,25 @@ this.getParameterValues = (task, methodResponse, cwmpResponse) ->
   task.nextIndex = Math.min(task.currentIndex + BATCH_SIZE, task.parameterNames.length)
   names = task.parameterNames.slice(task.currentIndex, task.nextIndex)
 
-  if names.length > 0
+  if methodResponse.type is 'GetParameterValuesResponse'
+    deviceUpdates.parameterValues = methodResponse.parameterList
+
+  if names.length == 0
+    return STATUS_FINISHED
+
   cwmpResponse.methodRequest = {
     type : 'GetParameterValues',
     parameterNames : names
   }
 
-  if methodResponse.type is 'GetParameterValuesResponse'
+  return STATUS_STARTED
-    return {'parameterValues' : methodResponse.parameterList}
+
-  return
+
+this.setParameterValues = (task, methodResponse, cwmpResponse, deviceUpdates) ->
+  if methodResponse.faultcode?
+    task.fault = methodResponse
+    return STATUS_FAULT
 
-this.setParameterValues = (task, methodResponse, cwmpResponse) ->
   if not task.currentIndex?
     task.currentIndex = 0
   else if methodResponse.type is 'SetParameterValuesResponse'
@@ -60,29 +95,48 @@ this.setParameterValues = (task, methodResponse, cwmpResponse) ->
   task.nextIndex = Math.min(task.currentIndex + BATCH_SIZE, task.parameterValues.length)
   values = task.parameterValues.slice(task.currentIndex, task.nextIndex)
 
-  if values.length > 0
+  if prevValues?
+    deviceUpdates.parameterValues = prevValues
+
+  if values.length == 0
+    return STATUS_FINISHED
+
   cwmpResponse.methodRequest = {
     type : 'SetParameterValues',
     parameterList : values
   }
+  return STATUS_STARTED
 
-  if prevValues?
-    return {'parameterValues' : prevValues}
-  return
 
-this.reboot = (task, methodResponse, cwmpResponse) ->
+this.reboot = (task, methodResponse, cwmpResponse, deviceUpdates) ->
+  if methodResponse.faultcode?
+    task.fault = methodResponse
+    return STATUS_FAULT
+
   if methodResponse.type isnt 'RebootResponse'
     cwmpResponse.methodRequest = {
       type : 'Reboot'
     }
+    return STATUS_STARTED
+
+  return STATUS_FINISHED
+
+
+this.download = (task, methodResponse, cwmpResponse, deviceUpdates) ->
+  if methodResponse.faultcode?
+    task.fault = methodResponse
+    return STATUS_FAULT
 
-this.download = (task, methodResponse, cwmpResponse) ->
   if methodResponse.type isnt 'DownloadResponse'
     cwmpResponse.methodRequest = {
       type : 'Download',
       fileType : task.fileType,
       url : task.url
     }
+    return STATUS_STARTED
 
-exports.task = (task, methodResponse, cwmpResponse) ->
+  return STATUS_FINISHED
-  return this[task.name](task, methodResponse, cwmpResponse)
+
+
+exports.task = (task, methodResponse, cwmpResponse, deviceUpdates) ->
+  return this[task.name](task, methodResponse, cwmpResponse, deviceUpdates)