swarm/storage: Perform CleanGCIndex iteration in chunks

swarm/storage/ldbstore.go

@@ -602,44 +602,76 @@ func (s *LDBStore) CleanGCIndex() error {
 		return err
 	}
 
-	it.Seek([]byte{keyIndex})
+	it.Release()
-	var idx dpaDBIndex
-	var poPtrs [256]uint64
-	for it.Valid() {
-		rowType, chunkHash := parseGCIdxKey(it.Key())
-		if rowType != keyIndex {
-			break
-		}
-		err := decodeIndex(it.Value(), &idx)
-		if err != nil {
-			return fmt.Errorf("corrupt index: %v", err)
-		}
-		po := s.po(chunkHash)
 
-		// if we don't find the data key, remove the entry
+	//var idx dpaDBIndex
-		dataKey := getDataKey(idx.Idx, po)
+	var poPtrs [256]uint64
-		_, err = s.db.Get(dataKey)
+	var doneIterating bool
-		if err != nil {
+	for !doneIterating {
-			log.Warn("deleting inconsistent index (missing data)", "key", chunkHash)
+		var idxs []dpaDBIndex
-			batch.Delete(it.Key())
+		var chunkHashes [][]byte
-		} else {
+		var pos []uint8
-			gcIdxKey := getGCIdxKey(&idx)
+		it := s.db.NewIterator()
-			gcIdxData := getGCIdxValue(&idx, po, chunkHash)
+		it.Seek([]byte{keyIndex})
-			batch.Put(gcIdxKey, gcIdxData)
+		for i := 0; i < 4096; i++ {
-			log.Trace("clean ok", "key", chunkHash, "gcKey", gcIdxKey, "gcData", gcIdxData)
+			if !it.Valid() {
-			okEntryCount++
+				doneIterating = true
-			if idx.Idx > poPtrs[po] {
+				break
-				poPtrs[po] = idx.Idx
 			}
+			rowType, chunkHash := parseGCIdxKey(it.Key())
+			if rowType != keyIndex {
+				doneIterating = true
+				break
+			}
+			//err := decodeIndex(it.Value(), &idx)
+			var idx dpaDBIndex
+			err := decodeIndex(it.Value(), &idx)
+			if err != nil {
+				return fmt.Errorf("corrupt index: %v", err)
+			}
+			po := s.po(chunkHash)
+
+			// if we don't find the data key, remove the entry
+			dataKey := getDataKey(idx.Idx, po)
+			_, err = s.db.Get(dataKey)
+			if err != nil {
+				log.Warn("deleting inconsistent index (missing data)", "key", chunkHash)
+				batch.Delete(it.Key())
+				//				if err := s.db.Delete(it.Key()); err != nil {
+				//					return err
+				//				}
+			} else {
+				idxs = append(idxs, idx)
+				chunkHashes = append(chunkHashes, chunkHash)
+				pos = append(pos, po)
+				okEntryCount++
+				if idx.Idx > poPtrs[po] {
+					poPtrs[po] = idx.Idx
+				}
+			}
+			totalEntryCount++
+			it.Next()
 		}
-		totalEntryCount++
+		it.Release()
-		if err := s.db.Write(&batch); err != nil {
+
+		err := s.db.Write(&batch)
+		if err != nil {
 			return err
 		}
-		it.Next()
+
+		for i, okIdx := range idxs {
+			gcIdxKey := getGCIdxKey(&okIdx)
+			gcIdxData := getGCIdxValue(&okIdx, pos[i], chunkHashes[i])
+			batch.Put(gcIdxKey, gcIdxData)
+			log.Trace("clean ok", "key", chunkHashes[i], "gcKey", gcIdxKey, "gcData", gcIdxData)
+		}
+		err = s.db.Write(&batch)
+		if err != nil {
+			return err
+		}
+
 	}
 
-	it.Release()
 	log.Debug("gc cleanup entries", "ok", okEntryCount, "total", totalEntryCount, "batchlen", batch.Len())
 
 	var entryCount [8]byte

swarm/storage/ldbstore_test.go

@@ -705,7 +705,7 @@ func TestCleanIndex(t *testing.T) {
 
 	// second gc index should still be fixed
 	if _, err := ldb.db.Get(gcSecondCorrectKey); err != nil {
-		t.Fatalf("expected gc 1 idx to be present: %v", idxKey)
+		t.Fatalf("expected gc 1 idx %v to be present: %v", gcSecondCorrectKey, idxKey)
 	}
 
 	// third gc index should be unchanged