New and improved fault management

This introduces the concept of "channel". Presets may belong to different
channels such that if one preset causes a fault, only presets belonging
to the same channel will be paused. Presets in other channels will continue
to be applied.

lib/api-functions.coffee

@@ -242,8 +242,9 @@ deleteDevice = (deviceId, callback) ->
     return callback(err) if err
     db.devicesCollection.remove({'_id' : deviceId}, (err) ->
       return callback(err) if err
-      db.redisClient.del("#{deviceId}_presets_hash", "#{deviceId}_inform_hash", (err) ->
-        callback(err)
+      db.faultsCollection.remove({'_id' : {'$regex' : "^#{common.escapeRegExp(deviceId)}\\:"}}, (err) ->
+        return callback(err) if err
+        db.redisClient.del("#{deviceId}_presets_hash", "#{deviceId}_inform_hash", "#{deviceId}_faults", "#{deviceId}_no_tasks", callback)
       )
     )
   )

lib/cache.coffee

@@ -141,7 +141,7 @@ refresh = (callback) ->
               else
                 throw new Error("Unknown configuration type #{c.type}")
 
-          presets.push({name: preset._id, schedule: schedule, events: events, precondition: precondition, provisions: _provisions})
+          presets.push({name: preset._id, channel: preset.channel or 'default', schedule: schedule, events: events, precondition: precondition, provisions: _provisions})
 
         if -- counter == 0
           computeHash()

lib/common.coffee

@@ -294,6 +294,11 @@ hammingWeight = (flags) ->
   return flags & 0x0000003f
 
 
+# Source: http://stackoverflow.com/a/6969486
+escapeRegExp = (str) ->
+  return str.replace(/[\-\[\]\/\{\}\(\)\*\+\?\.\\\^\$\|]/g, "\\$&")
+
+
 exports.UNDEFINED_TYPE = UNDEFINED_TYPE
 exports.NULL_TYPE = NULL_TYPE
 exports.NUMBER_TYPE = NUMBER_TYPE
@@ -307,3 +312,4 @@ exports.pathOverlap = pathOverlap
 exports.parsePath = parsePath
 exports.addPathMeta = addPathMeta
 exports.hammingWeight = hammingWeight
+exports.escapeRegExp = escapeRegExp

lib/cwmp.coffee

@@ -71,6 +71,7 @@ writeResponse = (currentRequest, res) ->
 
   if compress?
     compress(res.data, (err, data) ->
+      throw err if err
       res.headers['Content-Length'] = data.length
       currentRequest.httpResponse.writeHead(res.code, res.headers)
       currentRequest.httpResponse.end(data)
@@ -131,6 +132,28 @@ applyPresets = (currentRequest) ->
   cache.getPresets((err, presetsHash, presets) ->
     throw err if err
 
+    # Filter presets based on existing faults
+    blackList = null
+    whiteList = null
+    whiteListProvisions = null
+
+    RETRY_DELAY = config.get('RETRY_DELAY', currentRequest.sessionData.deviceId)
+
+    if currentRequest.sessionData.faults?
+      for channel, fault of currentRequest.sessionData.faults
+        if fault.retries?
+          retryTimestamp = fault.timestamp + (RETRY_DELAY * Math.pow(2, fault.retries)) * 1000
+        else
+          retryTimestamp = 0
+
+        if retryTimestamp <= currentRequest.sessionData.timestamp
+          whiteList = channel
+          whiteListProvisions = fault.provisions
+          break
+
+        blackList ?= []
+        blackList.push(channel)
+
     currentRequest.sessionData.presetsHash = presetsHash
 
     deviceEvents = {}
@@ -141,7 +164,13 @@ applyPresets = (currentRequest) ->
 
     parameters = {}
     filteredPresets = []
+
     for p in presets
+      if whiteList?
+        continue if p.channel != whiteList
+      else if blackList and p.channel in blackList
+        continue
+
       eventsMatch = true
       for k, v of p.events
         if !v != !deviceEvents[k.replace(' ', '_')]
@@ -170,9 +199,12 @@ applyPresets = (currentRequest) ->
         if unpacked[0] and (vv = deviceData.values.value.get(unpacked[0]))?
           parameters[k] = vv[0]
 
+      if whiteList?
+        session.addProvisions(currentRequest.sessionData, whiteList, whiteListProvisions)
+
       for p in filteredPresets
         if query.test(parameters, p.precondition)
-          session.addProvisions(currentRequest.sessionData, p.provisions)
+          session.addProvisions(currentRequest.sessionData, p.channel, p.provisions)
 
       session.rpcRequest(currentRequest.sessionData, null, (err, id, rpcRequest) ->
         throw err if err
@@ -190,48 +222,45 @@ nextRpc = (currentRequest) ->
     if rpcRequest?
       return sendRpcRequest(currentRequest, id, rpcRequest)
 
-    doneTasks = null
-    for p in currentRequest.sessionData.provisions when p[0] == '_task'
-      doneTasks ?= []
-      doneTasks.push(mongodb.ObjectID(p[1]))
+    for p, i in currentRequest.sessionData.provisions
+      if currentRequest.sessionData.faults?
+        delete currentRequest.sessionData.faults[currentRequest.sessionData.channels[i]]
+
+      if p[0] == '_task'
+        currentRequest.sessionData.doneTasks ?= []
+        currentRequest.sessionData.doneTasks.push(p[1])
+        for t, j in currentRequest.sessionData.tasks
+          if String(t._id) == p[1]
+            currentRequest.sessionData.tasks.splice(j, 1)
+            break
 
     session.clearProvisions(currentRequest.sessionData)
 
-    # No need to query for pending tasks if there was none in previous cycle
-    if not doneTasks? and currentRequest.sessionData.provisions?.length
-      return applyPresets(currentRequest)
-
-    nextTask = () ->
-      cur = db.tasksCollection.find({'device' : currentRequest.sessionData.deviceId, timestamp : {$lte : new Date(currentRequest.sessionData.timestamp)}}).sort(['timestamp']).limit(1)
-      cur.nextObject((err, task) ->
-        if not task?
-          return applyPresets(currentRequest)
+    for task in currentRequest.sessionData.tasks
+      # Delete if expired
+      if task.expiry <= currentRequest.sessionData.timestamp
+        util.log("#{currentRequest.sessionData.deviceId}: Task is expired #{task.name}(#{task._id})")
+        currentRequest.sessionData.doneTasks.push(String(task._id))
+        continue
 
       switch task.name
         when 'getParameterValues'
-            session.addProvisions(currentRequest.sessionData, [['_task', task._id, 'getParameterValues'].concat(task.parameterNames)])
+          session.addProvisions(currentRequest.sessionData, "_task_#{task._id}", [['_task', task._id, 'getParameterValues'].concat(task.parameterNames)])
         when 'setParameterValues'
           t = ['_task', task._id, 'setParameterValues']
           for p in task.parameterValues
             t.push(p[0])
             t.push(p[1])
             t.push(p[2] ? '')
-            session.addProvisions(currentRequest.sessionData, [t])
+          session.addProvisions(currentRequest.sessionData, "_task_#{task._id}", [t])
         when 'refreshObject'
-            session.addProvisions(currentRequest.sessionData, [['_task', task._id, 'refreshObject', task.objectName]])
+          session.addProvisions(currentRequest.sessionData, "_task_#{task._id}", [['_task', task._id, 'refreshObject', task.objectName]])
         else
           throw new Error('Task name not recognized')
 
       return nextRpc(currentRequest)
-      )
 
-    if not doneTasks?
-      return nextTask()
-
-    db.tasksCollection.remove({'_id' : {'$in' : doneTasks}}, (err, res) ->
-      throw err if err
-      return nextTask()
-    )
+    return applyPresets(currentRequest)
   )
 
 
@@ -242,8 +271,18 @@ sendRpcRequest = (currentRequest, id, rpcRequest) ->
       if isNew
         util.log("#{currentRequest.sessionData.deviceId}: New device registered")
 
+      db.clearTasks(currentRequest.sessionData.doneTasks, (err) ->
+        throw err if err
+
+        if not currentRequest.sessionData.faults?
+          return writeResponse(currentRequest, soap.response(null))
+
+        db.syncFaults(currentRequest.sessionData.deviceId, currentRequest.sessionData.faults, (err) ->
+          throw err if err
           writeResponse(currentRequest, soap.response(null))
         )
+      )
+    )
     return
 
   util.log("#{currentRequest.sessionData.deviceId}: #{rpcRequest.type} (#{id})")
@@ -324,9 +363,21 @@ listener = (httpRequest, httpResponse) ->
         deviceId = common.generateDeviceId(cwmpRequest.methodRequest.deviceId)
         return session.init(deviceId, cwmpRequest.cwmpVersion, cwmpRequest.sessionTimeout ? config.get('SESSION_TIMEOUT', deviceId), (err, sessionData) ->
           throw err if err
+
           httpRequest.connection.setTimeout(sessionData.timeout * 1000)
+
+          db.getDueTasksAndFaults(deviceId, sessionData.timestamp, (err, dueTasks, faults) ->
+            throw err if err
+            sessionData.tasks = dueTasks
+            sessionData.faults = faults if Object.keys(faults).length
+
+            # Delete expired faults
+            for k, v of sessionData.faults
+              delete sessionData.faults[k] if v.expiry >= sessionData.timestamp
+
             f(null, sessionData)
           )
+        )
 
       currentRequest = {
         httpRequest : httpRequest,
@@ -362,7 +413,50 @@ listener = (httpRequest, httpResponse) ->
           nextRpc(currentRequest)
         )
       else if cwmpRequest.fault?
-        session.rpcFault(sessionData, cwmpRequest.id, cwmpRequest.fault, (err) ->
+        session.rpcFault(currentRequest.sessionData, cwmpRequest.id, cwmpRequest.fault, (err, flt) ->
+          throw err if err
+
+          if not flt?
+            return nextRpc(currentRequest)
+
+          if not currentRequest.sessionData.provisions.length
+            throw new Error('A fault occured while trying to discover parameters to test preset preconditions: ' + JSON.stringify(flt))
+
+          faults = {}
+          currentRequest.sessionData.faults ?= {}
+
+          for p, i in currentRequest.sessionData.provisions
+            channel = currentRequest.sessionData.channels[i]
+            fault = faults[channel]
+            if not fault?
+              fault = {
+                timestamp: currentRequest.sessionData.timestamp
+                provisions: []
+                fault: flt
+              }
+
+              if p[0] == '_task'
+                for t, j in currentRequest.sessionData.tasks
+                  if t._id == p[1]
+                    currentRequest.sessionData.tasks.splice(j, 1)
+                    fault.expiry = t.expiry if t.expiry?
+                    break
+
+              faults[channel] = fault
+
+            fault.provisions.push(p)
+
+          for channel, fault of faults
+            if currentRequest.sessionData.faults[channel]?
+              fault.retries = (currentRequest.sessionData.faults[channel].retries ? 0) + 1
+
+            if not fault.retries? and Object.keys(faults).length == 1
+              fault.retries = 0
+
+            currentRequest.sessionData.faults[channel] = fault
+            util.log("#{currentRequest.sessionData.deviceId}: Fault response for channel #{channel} (retries: #{fault.retries ? 0})")
+
+          session.clearProvisions(currentRequest.sessionData)
           nextRpc(currentRequest)
         )
       else

lib/db.coffee

@@ -28,10 +28,11 @@ presetsCollection = null
 objectsCollection = null
 provisionsCollection = null
 virtualParametersCollection = null
+faultsCollection = null
 
 
 connect = (callback) ->
-  callbackCounter = 8
+  callbackCounter = 9
   mongodb.MongoClient.connect(config.get('MONGODB_CONNECTION_URL'), {db:{w:1},server:{autoReconnect:true}}, (err, db) ->
     return callback(err) if err
     exports.mongoDb = db
@@ -87,6 +88,13 @@ connect = (callback) ->
         return callback(err)
     )
 
+    db.collection('faults', (err, collection) ->
+      exports.faultsCollection = faultsCollection = collection
+      if --callbackCounter == 0 or err
+        callbackCounter = 0
+        return callback(err)
+    )
+
     exports.redisClient = redisClient = redis.createClient(config.get('REDIS_PORT'), config.get('REDIS_HOST'))
     redisClient.select(config.get('REDIS_DB'), (err) ->
       if --callbackCounter == 0 or err
@@ -101,19 +109,6 @@ disconnect = () ->
   redisClient.quit()
 
 
-getTask = (taskId, callback) ->
-  tid = String(taskId)
-  redisClient.get(tid, (err, res) ->
-    return callback(err) if err
-    if res?
-      callback(err, JSON.parse(res))
-    else
-      tasksCollection.findOne({_id : mongodb.ObjectID(tid)}, (err, task) ->
-        callback(err, task)
-      )
-  )
-
-
 getCached = (name, valueCallback, valueExpiry, callback) ->
   redisClient.get(name, (err, res) ->
     return callback(err) if err
@@ -549,10 +544,168 @@ saveDevice = (deviceId, deviceData, isNew, callback) ->
   )
 
 
-exports.getTask = getTask
+getDueTasksAndFaults = (deviceId, timestamp, callback) ->
+  redisClient.mget("#{deviceId}_no_tasks", "#{deviceId}_faults", (err, res) ->
+    return callback(err) if err
+
+    if res[0]?
+      tasks = []
+
+    if res[1]?
+      faults = JSON.parse(res[1])
+
+    if tasks? and faults?
+      return callback(null, tasks, faults)
+
+    CACHE_DURATION = config.get('PRESETS_CACHE_DURATION', deviceId)
+
+    if not faults?
+      getFaults(deviceId, (err, flts) ->
+        if err
+          callback?(err)
+          return callback = null
+
+        faults = flts
+        if tasks?
+          return callback(null, tasks, faults)
+      )
+
+    if not tasks?
+      getDueTasks(deviceId, timestamp, (err, dueTasks, nextTimestamp) ->
+        if err
+          callback?(err)
+          return callback = null
+
+        if nextTimestamp?
+          exp = Math.min(0, Math.trunc((nextTimestamp - Date.now()) / 1000))
+        else
+          exp = CACHE_DURATION
+
+        redisClient.setex("#{deviceId}_no_tasks", exp, 1, (err) ->
+          if err
+            callback?(err)
+            return callback = null
+
+          tasks = dueTasks
+          if faults?
+            return callback(null, tasks, faults)
+        )
+      )
+  )
+
+
+getFaults = (deviceId, callback) ->
+  faultsCollection.find({'_id' : {'$regex' : "^#{common.escapeRegExp(deviceId)}\\:"}}).toArray((err, res) ->
+    return callback(err) if err
+
+    faults = {}
+    for r in res
+      channel = r._id.slice(deviceId.length + 1)
+      delete r._id
+      r.timestamp = +r.timestamp
+      r.expiry = +r.expiry if r.expiry?
+      faults[channel] = r
+
+    CACHE_DURATION = config.get('PRESETS_CACHE_DURATION', deviceId)
+
+    redisClient.setex("#{deviceId}_faults", CACHE_DURATION, JSON.stringify(faults), (err) ->
+      callback(err, faults)
+    )
+  )
+
+
+
+getDueTasks = (deviceId, timestamp, callback) ->
+  cur = tasksCollection.find({
+    'device' : deviceId,
+    'fault' : {'$exists': false}
+    }).sort(['timestamp'])
+
+  tasks = []
+
+  cur.nextObject(f = (err, task) ->
+    return callback(err) if err
+
+    if not task?
+      return callback(null, tasks, null)
+
+    task.timestamp = +task.timestamp if task.timestamp?
+
+    if task.timestamp >= timestamp
+      return callback(null, tasks, +task.timestamp)
+
+    task._id = String(task._id)
+    tasks.push(task)
+    cur.nextObject(f)
+  )
+
+
+clearTasks = (taskIds, callback) ->
+  if not taskIds?.length
+    return callback()
+
+  tasksCollection.remove({'_id' : {'$in' : (mongodb.ObjectID(id) for id in taskIds)}}, callback)
+
+
+syncFaults = (deviceId, faults, callback) ->
+  getFaults(deviceId, (err, existingFaults) ->
+    return callback(err) if err
+
+    counter = 1
+    toDelete = []
+    for k of existingFaults
+      if k not of faults
+        toDelete.push("#{deviceId}:#{k}")
+
+    if toDelete.length
+      ++ counter
+      faultsCollection.remove({'_id' : {'$in' : toDelete}}, (err) ->
+        if err
+          callback(err) if counter
+          return counter = 0
+
+        if -- counter == 0
+          return redisClient.del("#{deviceId}_faults", callback)
+      )
+
+    for channel, fault of faults
+      continue if fault.retries == existingFaults[channel]?.retries
+      do (channel, fault) ->
+        ++ counter
+        fault._id = "#{deviceId}:#{channel}"
+        fault.timestamp = new Date(fault.timestamp)
+        fault.expiry = new Date(fault.expiry) if fault.expiry?
+
+        faultsCollection.save(fault, (err) ->
+          if err
+            callback(err) if counter
+            return counter = 0
+
+          if channel.startsWith('_task_')
+            return tasksCollection.update({_id: mongodb.ObjectID(channel.slice(6))}, {$set: {fault: fault.fault, retries: fault.retries}}, (err) ->
+              if err
+                callback(err) if counter
+                return counter = 0
+
+              if -- counter == 0
+                return redisClient.del("#{deviceId}_faults", callback)
+            )
+          else
+            if -- counter == 0
+              return redisClient.del("#{deviceId}_faults", callback)
+        )
+
+    if -- counter == 0
+      return callback()
+  )
+
+
 exports.getPresetsObjectsAliases = getPresetsObjectsAliases
 exports.getAliases = getAliases
 exports.connect = connect
 exports.disconnect = disconnect
 exports.fetchDevice = fetchDevice
 exports.saveDevice = saveDevice
+exports.getDueTasksAndFaults = getDueTasksAndFaults
+exports.clearTasks = clearTasks
+exports.syncFaults = syncFaults

lib/nbi.coffee

@@ -94,7 +94,7 @@ listener = (request, response) ->
         preset._id = presetName
 
         db.presetsCollection.save(preset, (err) ->
-          db.redisClient.del('presets', 'presets_hash', (err) ->
+          db.redisClient.del('presets_hash', (err) ->
             throw err if err
           )
           if err
@@ -106,7 +106,7 @@ listener = (request, response) ->
         )
       else if request.method == 'DELETE'
         db.presetsCollection.remove({'_id' : presetName}, (err) ->
-          db.redisClient.del('presets', 'presets_hash', (err) ->
+          db.redisClient.del('presets_hash', (err) ->
             throw err if err
           )
           if err
@@ -258,13 +258,9 @@ listener = (request, response) ->
           task.device = deviceId
           db.getAliases((aliases) ->
             apiFunctions.insertTasks(task, aliases, (err) ->
-              db.redisClient.del("#{deviceId}_presets_hash", (err) ->
               throw err if err
-              )
-              if err
-                response.writeHead(500)
-                response.end(errorToString(err))
-                return
+              db.redisClient.del("#{deviceId}_no_tasks", (err) ->
+                throw err if err
 
                 if urlParts.query.connection_request?
                   apiFunctions.connectionRequest(deviceId, (err) ->
@@ -301,6 +297,7 @@ listener = (request, response) ->
                   response.end(JSON.stringify(task))
               )
             )
+          )
         else if urlParts.query.connection_request?
           # no task, send connection request only
           apiFunctions.connectionRequest(deviceId, (err) ->
@@ -323,24 +320,38 @@ listener = (request, response) ->
       action = r[2]
       if not action? or action is '/'
         if request.method == 'DELETE'
+          db.tasksCollection.findOne({'_id' : taskId}, {'device' : 1}, (err, task) ->
+            throw err if err
+            deviceId = task.device
             db.tasksCollection.remove({'_id' : taskId}, (err) ->
-            if err
-              response.writeHead(500)
-              response.end(errorToString(err))
-              return
+              throw err if err
+              db.faultsCollection.remove({_id : "#{deviceId}:_task_#{String(taskId)}"}, (err) ->
+                throw err if err
                 response.writeHead(200)
                 response.end()
               )
+            )
+          )
         else
           response.writeHead 405, {'Allow': 'PUT DELETE'}
           response.end('405 Method Not Allowed')
       else if action is '/retry'
         if request.method == 'POST'
+          db.tasksCollection.findOne({'_id' : taskId}, {'device' : 1}, (err, task) ->
+            throw err if err
+            deviceId = task.device
             db.tasksCollection.update({_id : taskId}, {$unset : {fault : 1}, $set : {timestamp : new Date()}}, (err, count) ->
-            # TODO need to invalidate presets hash for the device
+              throw err if err
+              db.faultsCollection.remove({_id : "#{deviceId}:_task_#{String(taskId)}"}, (err) ->
+                throw err if err
+                db.redisClient.del("#{deviceId}_no_tasks", "#{deviceId}_faults", (err) ->
+                  throw err if err
                   response.writeHead(200)
                   response.end()
                 )
+              )
+            )
+          )
         else
           response.writeHead 405, {'Allow': 'POST'}
           response.end('405 Method Not Allowed')

lib/session.coffee

@@ -48,6 +48,7 @@ init = (deviceId, cwmpVersion, timeout, callback) ->
     cwmpVersion : cwmpVersion
     timeout : timeout
     provisions: []
+    channels: []
     revisions: [0]
     rpcCount: 0
     cycle: 0
@@ -188,8 +189,17 @@ inform = (sessionData, rpcReq, callback) ->
   )
 
 
-addProvisions = (sessionData, provisions) ->
-  sessionData.provisions = sessionData.provisions.concat(provisions)
+addProvisions = (sessionData, channel, provisions) ->
+  for provision, i in provisions
+    # Remove duplicate provisions
+    for p, j in sessionData.provisions
+      if channel == sessionData.channels[j] and JSON.stringify(p) == JSON.stringify(provision)
+        sessionData.provisions.splice(j, 1)
+        sessionData.channels.splice(j, 1)
+
+    sessionData.provisions.push(provision)
+    sessionData.channels.push(channel)
+
   if sessionData.revisions.length > 1 or sessionData.revisions[0] > 0
     v.collapse(1) for k, v of sessionData.timestamps
     v.collapse(1) for k, v of sessionData.values
@@ -200,7 +210,10 @@ addProvisions = (sessionData, provisions) ->
 
 
 clearProvisions = (sessionData) ->
+  delete sessionData.syncState
   sessionData.provisions = []
+  sessionData.channels = []
+  sessionData.declarations = []
   if sessionData.revisions.length > 1 or sessionData.revisions[0] > 0
     v.collapse(1) for k, v of sessionData.deviceData.timestamps
     v.collapse(1) for k, v of sessionData.deviceData.values
@@ -1064,7 +1077,10 @@ rpcResponse = (sessionData, id, rpcRes, callback) ->
 
 
 rpcFault = (sessionData, id, faultResponse, callback) ->
-  throw new Error('Not implemented')
+  # TODO Consider handling invalid parameter faults by automatically refreshing
+  # relevant data model portions
+  delete sessionData.rpcRequest
+  return callback(null, faultResponse)
 
 
 load = (id, callback) ->
@@ -1076,11 +1092,13 @@ load = (id, callback) ->
 
       sessionData = JSON.parse(sessionDataString)
 
-      if sessionData.cache.hash != hash
+      if sessionData.presetsHash? and sessionData.presetsHash != hash
         return callback(new Error('Preset hash mismatch'))
 
-      sessionData.cache.provisions = provisions
-      sessionData.cache.virtualParameters = virtualParameters
+      sessionData.cache = {
+        provisions: provisions
+        virtualParameters: virtualParameters
+      }
 
       for d in sessionData.declarations
         common.addPathMeta(d[0])
@@ -1110,8 +1128,7 @@ load = (id, callback) ->
 save = (sessionData, callback) ->
   delete sessionData.syncState
   delete sessionData.toLoad
-  delete sessionData.cache.provisions
-  delete sessionData.cache.virtualParameters
+  delete sessionData.cache
 
   sessionData.id ?= crypto.randomBytes(8).toString('hex')
   deviceData = []