Merge pull request #1403 from mrjana/networkdb

Ignore delete events for non-existent entries

networkdb/cluster.go

@@ -200,10 +200,7 @@ func (nDB *NetworkDB) reapNetworks() {
 }
 
 func (nDB *NetworkDB) reapTableEntries() {
-	var (
-		paths   []string
-		entries []*entry
-	)
+	var paths []string
 
 	now := time.Now()
 
@@ -219,14 +216,12 @@ func (nDB *NetworkDB) reapTableEntries() {
 		}
 
 		paths = append(paths, path)
-		entries = append(entries, entry)
 		return false
 	})
 	nDB.RUnlock()
 
 	nDB.Lock()
-	for i, path := range paths {
-		entry := entries[i]
+	for _, path := range paths {
 		params := strings.Split(path[1:], "/")
 		tname := params[0]
 		nid := params[1]
@@ -239,8 +234,6 @@ func (nDB *NetworkDB) reapTableEntries() {
 		if _, ok := nDB.indexes[byNetwork].Delete(fmt.Sprintf("/%s/%s/%s", nid, tname, key)); !ok {
 			logrus.Errorf("Could not delete entry in network %s with table name %s and key %s as it does not exist", nid, tname, key)
 		}
-
-		nDB.broadcaster.Write(makeEvent(opDelete, tname, nid, key, entry.value))
 	}
 	nDB.Unlock()
 }

networkdb/delegate.go

@@ -84,28 +84,34 @@ func (nDB *NetworkDB) handleTableEvent(tEvent *TableEvent) bool {
 		return true
 	}
 
-	if entry, err := nDB.getEntry(tEvent.TableName, tEvent.NetworkID, tEvent.Key); err == nil {
+	e, err := nDB.getEntry(tEvent.TableName, tEvent.NetworkID, tEvent.Key)
+	if err != nil && tEvent.Type == TableEventTypeDelete {
+		// If it is a delete event and we don't have the entry here nothing to do.
+		return false
+	}
+
+	if err == nil {
 		// We have the latest state. Ignore the event
 		// since it is stale.
-		if entry.ltime >= tEvent.LTime {
+		if e.ltime >= tEvent.LTime {
 			return false
 		}
 	}
 
-	entry := &entry{
+	e = &entry{
 		ltime:    tEvent.LTime,
 		node:     tEvent.NodeName,
 		value:    tEvent.Value,
 		deleting: tEvent.Type == TableEventTypeDelete,
 	}
 
-	if entry.deleting {
-		entry.deleteTime = time.Now()
+	if e.deleting {
+		e.deleteTime = time.Now()
 	}
 
 	nDB.Lock()
-	nDB.indexes[byTable].Insert(fmt.Sprintf("/%s/%s/%s", tEvent.TableName, tEvent.NetworkID, tEvent.Key), entry)
-	nDB.indexes[byNetwork].Insert(fmt.Sprintf("/%s/%s/%s", tEvent.NetworkID, tEvent.TableName, tEvent.Key), entry)
+	nDB.indexes[byTable].Insert(fmt.Sprintf("/%s/%s/%s", tEvent.TableName, tEvent.NetworkID, tEvent.Key), e)
+	nDB.indexes[byNetwork].Insert(fmt.Sprintf("/%s/%s/%s", tEvent.NetworkID, tEvent.TableName, tEvent.Key), e)
 	nDB.Unlock()
 
 	var op opType

networkdb/networkdb.go

@@ -326,6 +326,8 @@ func (nDB *NetworkDB) deleteNodeTableEntries(node string) {
 
 		nDB.indexes[byTable].Insert(fmt.Sprintf("/%s/%s/%s", tname, nid, key), entry)
 		nDB.indexes[byNetwork].Insert(fmt.Sprintf("/%s/%s/%s", nid, tname, key), entry)
+
+		nDB.broadcaster.Write(makeEvent(opDelete, tname, nid, key, entry.value))
 		return false
 	})
 	nDB.Unlock()