Merge pull request ipfs#1977 from ipfs/fix/record-accounting

send record fixes to peers who send outdated records

routing/dht/dht.go

@@ -150,6 +150,8 @@ func (dht *IpfsDHT) putProvider(ctx context.Context, p peer.ID, skey string) err
 	return nil
 }
 
+var errInvalidRecord = errors.New("received invalid record")
+
 // getValueOrPeers queries a particular peer p for the value for
 // key. It returns either the value or a list of closer peers.
 // NOTE: it will update the dht's peerstore with any new addresses
@@ -173,9 +175,11 @@ func (dht *IpfsDHT) getValueOrPeers(ctx context.Context, p peer.ID,
 		err = dht.verifyRecordOnline(ctx, record)
 		if err != nil {
 			log.Info("Received invalid record! (discarded)")
-			return nil, nil, err
+			// return a sentinal to signify an invalid record was received
+			err = errInvalidRecord
+			record = new(pb.Record)
 		}
-		return record, peers, nil
+		return record, peers, err
 	}
 
 	if len(peers) > 0 {

routing/dht/routing.go

@@ -91,7 +91,9 @@ func (dht *IpfsDHT) GetValue(ctx context.Context, key key.Key) ([]byte, error) {
 
 	var recs [][]byte
 	for _, v := range vals {
-		recs = append(recs, v.Val)
+		if v.Val != nil {
+			recs = append(recs, v.Val)
+		}
 	}
 
 	i, err := dht.Selector.BestRecord(key, recs)
@@ -169,13 +171,26 @@ func (dht *IpfsDHT) GetValues(ctx context.Context, key key.Key, nvals int) ([]ro
 		})
 
 		rec, peers, err := dht.getValueOrPeers(ctx, p, key)
-		if err != nil {
+		switch err {
+		case routing.ErrNotFound:
+			// in this case, they responded with nothing,
+			// still send a notification so listeners can know the
+			// request has completed 'successfully'
+			notif.PublishQueryEvent(parent, &notif.QueryEvent{
+				Type: notif.PeerResponse,
+				ID:   p,
+			})
+			return nil, err
+		default:
 			return nil, err
+
+		case nil, errInvalidRecord:
+			// in either of these cases, we want to keep going
 		}
 
 		res := &dhtQueryResult{closerPeers: peers}
 
-		if rec.GetValue() != nil {
+		if rec.GetValue() != nil || err == errInvalidRecord {
 			rv := routing.RecvdVal{
 				Val:  rec.GetValue(),
 				From: p,