Reduce memory consumption for prune operation

For non-exhaustive prune, consider only target chunks instead of mapping
all chunks in repository.

src/duplicacy_snapshotmanager.go

@@ -26,6 +26,7 @@ import (
 
 const (
 	secondsInDay = 86400
+	chunkDir     = "chunks/"
 )
 
 // FossilCollection contains fossils and temporary files found during a snapshot deletions.
@@ -455,7 +456,7 @@ func (manager *SnapshotManager) CleanSnapshotCache(latestSnapshot *Snapshot, all
 		}
 	}
 
-	allFiles, _ := manager.ListAllFiles(manager.snapshotCache, "chunks/")
+	allFiles, _ := manager.ListAllFiles(manager.snapshotCache, chunkDir)
 	for _, file := range allFiles {
 		if file[len(file)-1] != '/' {
 			chunkID := strings.Replace(file, "/", "", -1)
@@ -753,7 +754,7 @@ func (manager *SnapshotManager) CheckSnapshots(snapshotID string, revisionsToChe
 	chunkSnapshotMap := make(map[string]int)
 
 	LOG_INFO("SNAPSHOT_CHECK", "Listing all chunks")
-	allChunks, allSizes := manager.ListAllFiles(manager.storage, "chunks/")
+	allChunks, allSizes := manager.ListAllFiles(manager.storage, chunkDir)
 
 	for i, chunk := range allChunks {
 		if len(chunk) == 0 || chunk[len(chunk)-1] == '/' {
@@ -1590,15 +1591,24 @@ func (manager *SnapshotManager) PruneSnapshots(selfID string, snapshotID string,
 		LOG_WARN("DELETE_OPTIONS", "Tags or retention policy will be ignored if at least one revision is specified")
 	}
 
-	preferencePath := GetDuplicacyPreferencePath()
+	prefPath := GetDuplicacyPreferencePath()
-	logDir := path.Join(preferencePath, "logs")
+	logDir := path.Join(prefPath, "logs")
-	os.Mkdir(logDir, 0700)
+	err := os.MkdirAll(logDir, 0700)
+	if err != nil {
+		LOG_ERROR("LOG_DIR", "Could not open log directory %s: %v", logDir, err)
+	}
 	logFileName := path.Join(logDir, time.Now().Format("prune-log-20060102-150405"))
 	logFile, err := os.OpenFile(logFileName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0600)
+	if err != nil {
+		LOG_ERROR("LOG_FILE", "Could not open log file %s: %v", logFileName, err)
+	}
 
 	defer func() {
 		if logFile != nil {
-			logFile.Close()
+			cerr := logFile.Close()
+			if cerr != nil {
+				LOG_WARN("LOG_FILE", "Could not close log file %s: %v", logFileName, cerr)
+			}
 		}
 	}()
 
@@ -1677,7 +1687,8 @@ func (manager *SnapshotManager) PruneSnapshots(selfID string, snapshotID string,
 	}
 
 	for _, id := range snapshotIDs {
-		revisions, err := manager.ListSnapshotRevisions(id)
+		var revisions []int
+		revisions, err = manager.ListSnapshotRevisions(id)
 		if err != nil {
 			LOG_ERROR("SNAPSHOT_LIST", "Failed to list all revisions for snapshot %s: %v", id, err)
 			return false
@@ -1697,14 +1708,20 @@ func (manager *SnapshotManager) PruneSnapshots(selfID string, snapshotID string,
 		}
 	}
 
-	chunkDir := "chunks/"
-
 	collectionRegex := regexp.MustCompile(`^([0-9]+)$`)
 
 	collectionDir := "fossils"
-	manager.snapshotCache.CreateDirectory(0, collectionDir)
+	err = manager.snapshotCache.CreateDirectory(0, collectionDir)
+	if err != nil {
+		LOG_ERROR("FOSSIL_COLLECT", "Failed to create collection directory %s: %v", collectionDir, err)
+		return false
+	}
 
 	collections, _, err := manager.snapshotCache.ListFiles(0, collectionDir)
+	if err != nil {
+		LOG_ERROR("FOSSIL_COLLECT", "Failed to list fossil collection files for dir %s: %v", collectionDir, err)
+		return false
+	}
 	maxCollectionNumber := 0
 
 	referencedFossils := make(map[string]bool)
@@ -1730,7 +1747,7 @@ func (manager *SnapshotManager) PruneSnapshots(selfID string, snapshotID string,
 		collectionFile := path.Join(collectionDir, collectionName)
 		manager.fileChunk.Reset(false)
 
-		err := manager.snapshotCache.DownloadFile(0, collectionFile, manager.fileChunk)
+		err = manager.snapshotCache.DownloadFile(0, collectionFile, manager.fileChunk)
 		if err != nil {
 			LOG_ERROR("FOSSIL_COLLECT", "Failed to read the fossil collection file %s: %v", collectionFile, err)
 			return false
@@ -1784,12 +1801,16 @@ func (manager *SnapshotManager) PruneSnapshots(selfID string, snapshotID string,
 					if dryRun {
 						LOG_INFO("FOSSIL_DELETE", "The chunk %s would be permanently removed", chunk)
 					} else {
-						manager.storage.DeleteFile(0, fossil)
+						err = manager.storage.DeleteFile(0, fossil)
+						if err != nil {
+							LOG_WARN("FOSSIL_DELETE", "The chunk %s could not be removed: %v", chunk, err)
+						} else {
 							LOG_INFO("FOSSIL_DELETE", "The chunk %s has been permanently removed", chunk)
 							fmt.Fprintf(logFile, "Deleted fossil %s (collection %s)\n", chunk, collectionName)
 						}
 					}
 				}
+			}
 
 			// Delete all temporary files if they still exist.
 			for _, temporary := range collection.Temporaries {
@@ -1797,7 +1818,7 @@ func (manager *SnapshotManager) PruneSnapshots(selfID string, snapshotID string,
 					LOG_INFO("TEMPORARY_DELETE", "The temporary file %s would be deleted", temporary)
 				} else {
 					// Fail silently, since temporary files are supposed to be renamed or deleted after upload is done
-					manager.storage.DeleteFile(0, temporary)
+					_ = manager.storage.DeleteFile(0, temporary)
 					LOG_INFO("TEMPORARY_DELETE", "The temporary file %s has been deleted", temporary)
 					fmt.Fprintf(logFile, "Deleted temporary %s (collection %s)\n", temporary, collectionName)
 				}
@@ -1915,208 +1936,32 @@ func (manager *SnapshotManager) PruneSnapshots(selfID string, snapshotID string,
 		}
 	}
 
-	if toBeDeleted == 0 && exhaustive == false {
+	if toBeDeleted == 0 && !exhaustive {
 		LOG_INFO("SNAPSHOT_NONE", "No snapshot to delete")
 		return false
 	}
 
-	chunkRegex := regexp.MustCompile(`^[0-9a-f]+$`)
-
-	referencedChunks := make(map[string]bool)
-
-	// Now build all chunks referened by snapshot not deleted
-	for _, snapshots := range allSnapshots {
-
-		if len(snapshots) > 0 {
-			latest := snapshots[len(snapshots)-1]
-			if latest.Flag && !exclusive {
-				LOG_ERROR("SNAPSHOT_DELETE",
-					"The latest snapshot %s at revision %d can't be deleted in non-exclusive mode",
-					latest.ID, latest.Revision)
-				return false
-			}
-		}
-
-		for _, snapshot := range snapshots {
-			if snapshot.Flag {
-				LOG_INFO("SNAPSHOT_DELETE", "Deleting snapshot %s at revision %d", snapshot.ID, snapshot.Revision)
-				continue
-			}
-
-			chunks := manager.GetSnapshotChunks(snapshot, false)
-
-			for _, chunk := range chunks {
-				// The initial value is 'false'.  When a referenced chunk is found it will change the value to 'true'.
-				referencedChunks[chunk] = false
-			}
-		}
-	}
-
 	collection := CreateFossilCollection(allSnapshots)
 
+	var success bool
 	if exhaustive {
-
+		success = manager.pruneSnapshotsExhaustive(referencedFossils, allSnapshots, collection, logFile, dryRun, exclusive)
-		// In exhaustive, we scan the entire chunk tree to find dangling chunks and temporaries.
+	} else {
-		allFiles, _ := manager.ListAllFiles(manager.storage, chunkDir)
+		success = manager.pruneSnapshots(allSnapshots, collection, logFile, dryRun, exclusive)
-		for _, file := range allFiles {
-			if file[len(file)-1] == '/' {
-				continue
 	}
-
+	if !success {
-			if strings.HasSuffix(file, ".tmp") {
-
-				// This is a temporary chunk file.  It can be a result of a restore operation still in progress, or
-				// a left-over from a restore operation that was terminated abruptly.
-				if dryRun {
-					LOG_INFO("CHUNK_TEMPORARY", "Found temporary file %s", file)
-					continue
-				}
-
-				if exclusive {
-					// In exclusive mode, we assume no other restore operation is running concurrently.
-					err := manager.storage.DeleteFile(0, chunkDir+file)
-					if err != nil {
-						LOG_ERROR("CHUNK_TEMPORARY", "Failed to remove the temporary file %s: %v", file, err)
 		return false
-					} else {
-						LOG_DEBUG("CHUNK_TEMPORARY", "Deleted temporary file %s", file)
-					}
-					fmt.Fprintf(logFile, "Deleted temporary %s\n", file)
-				} else {
-					collection.AddTemporary(file)
-				}
-				continue
-			} else if strings.HasSuffix(file, ".fsl") {
-				// This is a fossil.  If it is unreferenced, it can be a result of failing to save the fossil
-				// collection file after making it a fossil.
-				if _, found := referencedFossils[file]; !found {
-					if dryRun {
-						LOG_INFO("FOSSIL_UNREFERENCED", "Found unreferenced fossil %s", file)
-						continue
-					}
-
-					chunk := strings.Replace(file, "/", "", -1)
-					chunk = strings.Replace(chunk, ".fsl", "", -1)
-
-					if _, found := referencedChunks[chunk]; found {
-						manager.resurrectChunk(chunkDir+file, chunk)
-					} else {
-						err := manager.storage.DeleteFile(0, chunkDir+file)
-						if err != nil {
-							LOG_WARN("FOSSIL_DELETE", "Failed to remove the unreferenced fossil %s: %v", file, err)
-						} else {
-							LOG_DEBUG("FOSSIL_DELETE", "Deleted unreferenced fossil %s", file)
-						}
-						fmt.Fprintf(logFile, "Deleted unreferenced fossil %s\n", file)
-					}
-				}
-
-				continue
-			}
-
-			chunk := strings.Replace(file, "/", "", -1)
-
-			if !chunkRegex.MatchString(chunk) {
-				LOG_WARN("CHUNK_UNKONWN_FILE", "File %s is not a chunk", file)
-				continue
-			}
-
-			if value, found := referencedChunks[chunk]; !found {
-
-				if dryRun {
-					LOG_INFO("CHUNK_UNREFERENCED", "Found unreferenced chunk %s", chunk)
-					continue
-				}
-
-				manager.fossilizeChunk(chunk, chunkDir+file, exclusive, collection)
-				if exclusive {
-					fmt.Fprintf(logFile, "Deleted chunk %s (exclusive mode)\n", chunk)
-				} else {
-					fmt.Fprintf(logFile, "Marked fossil %s\n", chunk)
-				}
-
-			} else if value {
-
-				// Note that the initial value is false.  So if the value is true it means another copy of the chunk
-				// exists in a higher-level directory.
-
-				if dryRun {
-					LOG_INFO("CHUNK_REDUNDANT", "Found redundant chunk %s", chunk)
-					continue
-				}
-
-				// This is a redundant chunk file (for instance D3/495A8D and D3/49/5A8D )
-				err := manager.storage.DeleteFile(0, chunkDir+file)
-				if err != nil {
-					LOG_WARN("CHUNK_DELETE", "Failed to remove the redundant chunk file %s: %v", file, err)
-				} else {
-					LOG_TRACE("CHUNK_DELETE", "Removed the redundant chunk file %s", file)
-				}
-				fmt.Fprintf(logFile, "Deleted redundant chunk %s\n", file)
-
-			} else {
-				referencedChunks[chunk] = true
-				LOG_DEBUG("CHUNK_KEEP", "Chunk %s is referenced", chunk)
-			}
-		}
-	} else {
-		// In non-exhaustive mode, only chunks that exist in the snapshots to be deleted but not other are identified
-		// as unreferenced chunks.
-		for _, snapshots := range allSnapshots {
-			for _, snapshot := range snapshots {
-
-				if !snapshot.Flag {
-					continue
-				}
-
-				chunks := manager.GetSnapshotChunks(snapshot, false)
-
-				for _, chunk := range chunks {
-
-					if _, found := referencedChunks[chunk]; found {
-						continue
-					}
-
-					if dryRun {
-						LOG_INFO("CHUNK_UNREFERENCED", "Found unreferenced chunk %s", chunk)
-						continue
-					}
-
-					chunkPath, exist, _, err := manager.storage.FindChunk(0, chunk, false)
-					if err != nil {
-						LOG_ERROR("CHUNK_FIND", "Failed to locate the path for the chunk %s: %v", chunk, err)
-						return false
-					}
-
-					if !exist {
-						LOG_WARN("CHUNK_MISSING", "The chunk %s referenced by snapshot %s revision %d does not exist",
-							chunk, snapshot.ID, snapshot.Revision)
-						continue
-					}
-
-					manager.fossilizeChunk(chunk, chunkPath, exclusive, collection)
-					if exclusive {
-						fmt.Fprintf(logFile, "Deleted chunk %s (exclusive mode)\n", chunk)
-					} else {
-						fmt.Fprintf(logFile, "Marked fossil %s\n", chunk)
-					}
-
-					referencedChunks[chunk] = true
-				}
-			}
-		}
-
 	}
 
 	// Save the fossil collection if it is not empty.
 	if !collection.IsEmpty() && !dryRun {
-
 		collection.EndTime = time.Now().Unix()
 
 		collectionNumber := maxCollectionNumber + 1
 		collectionFile := path.Join(collectionDir, fmt.Sprintf("%d", collectionNumber))
 
-		description, err := json.Marshal(collection)
+		var description []byte
+		description, err = json.Marshal(collection)
 		if err != nil {
 			LOG_ERROR("FOSSIL_COLLECT", "Failed to create a json file for the fossil collection: %v", err)
 			return false
@@ -2145,21 +1990,27 @@ func (manager *SnapshotManager) PruneSnapshots(selfID string, snapshotID string,
 				LOG_ERROR("SNAPSHOT_DELETE", "Failed to delete the snapshot %s at revision %d: %v",
 					snapshot.ID, snapshot.Revision, err)
 				return false
-			} else {
+			}
 			LOG_INFO("SNAPSHOT_DELETE", "The snapshot %s at revision %d has been removed",
 				snapshot.ID, snapshot.Revision)
-			}
+			err = manager.snapshotCache.DeleteFile(0, snapshotPath)
-			manager.snapshotCache.DeleteFile(0, snapshotPath)
+			if err != nil {
+				LOG_WARN("SNAPSHOT_DELETE", "The cached snapshot %s at revision %d could not be removed: %v",
+					snapshot.ID, snapshot.Revision, err)
+				fmt.Fprintf(logFile, "Cached snapshot %s at revision %d could not be removed: %v",
+					snapshot.ID, snapshot.Revision, err)
+			} else {
 				fmt.Fprintf(logFile, "Deleted cached snapshot %s at revision %d\n", snapshot.ID, snapshot.Revision)
 			}
 		}
+	}
 
 	if collection.IsEmpty() && !dryRun && toBeDeleted != 0 && !exclusive {
 		LOG_INFO("FOSSIL_NONE",
 			"No fossil collection has been created since deleted snapshots did not reference any unique chunks")
 	}
 
-	var latestSnapshot *Snapshot = nil
+	var latestSnapshot *Snapshot
 	if len(allSnapshots[selfID]) > 0 {
 		latestSnapshot = allSnapshots[selfID][len(allSnapshots[selfID])-1]
 	}
@@ -2173,6 +2024,223 @@ func (manager *SnapshotManager) PruneSnapshots(selfID string, snapshotID string,
 	return true
 }
 
+// pruneSnapshots in non-exhaustive mode, only chunks that exist in the
+// snapshots to be deleted but not other are identified as unreferenced chunks.
+func (manager *SnapshotManager) pruneSnapshots(allSnapshots map[string][]*Snapshot, collection *FossilCollection, logFile io.Writer, dryRun, exclusive bool) bool {
+	targetChunks := make(map[string]bool)
+
+	// Now build all chunks referened by snapshot not deleted
+	for _, snapshots := range allSnapshots {
+
+		if len(snapshots) > 0 {
+			latest := snapshots[len(snapshots)-1]
+			if latest.Flag && !exclusive {
+				LOG_ERROR("SNAPSHOT_DELETE",
+					"The latest snapshot %s at revision %d can't be deleted in non-exclusive mode",
+					latest.ID, latest.Revision)
+				return false
+			}
+		}
+
+		for _, snapshot := range snapshots {
+			if !snapshot.Flag {
+				continue
+			}
+
+			LOG_INFO("SNAPSHOT_DELETE", "Deleting snapshot %s at revision %d", snapshot.ID, snapshot.Revision)
+			chunks := manager.GetSnapshotChunks(snapshot, false)
+
+			for _, chunk := range chunks {
+				// The initial value is 'false'.  When a referenced chunk is found it will change the value to 'true'.
+				targetChunks[chunk] = false
+			}
+		}
+	}
+
+	for _, snapshots := range allSnapshots {
+		for _, snapshot := range snapshots {
+			if snapshot.Flag {
+				continue
+			}
+
+			chunks := manager.GetSnapshotChunks(snapshot, false)
+
+			for _, chunk := range chunks {
+				if _, found := targetChunks[chunk]; found {
+					targetChunks[chunk] = true
+				}
+			}
+		}
+	}
+
+	for chunk, value := range targetChunks {
+		if value {
+			continue
+		}
+
+		if dryRun {
+			LOG_INFO("CHUNK_UNREFERENCED", "Found unreferenced chunk %s", chunk)
+			continue
+		}
+
+		chunkPath, exist, _, err := manager.storage.FindChunk(0, chunk, false)
+		if err != nil {
+			LOG_ERROR("CHUNK_FIND", "Failed to locate the path for the chunk %s: %v", chunk, err)
+			return false
+		}
+
+		if !exist {
+			LOG_WARN("CHUNK_MISSING", "The chunk %s does not exist", chunk)
+			continue
+		}
+
+		manager.fossilizeChunk(chunk, chunkPath, exclusive, collection)
+		if exclusive {
+			fmt.Fprintf(logFile, "Deleted chunk %s (exclusive mode)\n", chunk)
+		} else {
+			fmt.Fprintf(logFile, "Marked fossil %s\n", chunk)
+		}
+
+		targetChunks[chunk] = true
+	}
+
+	return true
+}
+
+// pruneSnapshotsExhaustive in exhaustive, we scan the entire chunk tree to
+// find dangling chunks and temporaries.
+func (manager *SnapshotManager) pruneSnapshotsExhaustive(referencedFossils map[string]bool, allSnapshots map[string][]*Snapshot, collection *FossilCollection, logFile io.Writer, dryRun, exclusive bool) bool {
+	chunkRegex := regexp.MustCompile(`^[0-9a-f]+$`)
+	referencedChunks := make(map[string]bool)
+
+	// Now build all chunks referened by snapshot not deleted
+	for _, snapshots := range allSnapshots {
+		if len(snapshots) > 0 {
+			latest := snapshots[len(snapshots)-1]
+			if latest.Flag && !exclusive {
+				LOG_ERROR("SNAPSHOT_DELETE",
+					"The latest snapshot %s at revision %d can't be deleted in non-exclusive mode",
+					latest.ID, latest.Revision)
+				return false
+			}
+		}
+
+		for _, snapshot := range snapshots {
+			if snapshot.Flag {
+				LOG_INFO("SNAPSHOT_DELETE", "Deleting snapshot %s at revision %d", snapshot.ID, snapshot.Revision)
+				continue
+			}
+
+			chunks := manager.GetSnapshotChunks(snapshot, false)
+
+			for _, chunk := range chunks {
+				// The initial value is 'false'.  When a referenced chunk is found it will change the value to 'true'.
+				referencedChunks[chunk] = false
+			}
+		}
+	}
+
+	allFiles, _ := manager.ListAllFiles(manager.storage, chunkDir)
+	for _, file := range allFiles {
+		if file[len(file)-1] == '/' {
+			continue
+		}
+
+		if strings.HasSuffix(file, ".tmp") {
+			// This is a temporary chunk file.  It can be a result of a restore operation still in progress, or
+			// a left-over from a restore operation that was terminated abruptly.
+			if dryRun {
+				LOG_INFO("CHUNK_TEMPORARY", "Found temporary file %s", file)
+				continue
+			}
+
+			if exclusive {
+				// In exclusive mode, we assume no other restore operation is running concurrently.
+				err := manager.storage.DeleteFile(0, chunkDir+file)
+				if err != nil {
+					LOG_ERROR("CHUNK_TEMPORARY", "Failed to remove the temporary file %s: %v", file, err)
+					return false
+				}
+				LOG_DEBUG("CHUNK_TEMPORARY", "Deleted temporary file %s", file)
+				fmt.Fprintf(logFile, "Deleted temporary %s\n", file)
+			} else {
+				collection.AddTemporary(file)
+			}
+			continue
+		} else if strings.HasSuffix(file, ".fsl") {
+			// This is a fossil.  If it is unreferenced, it can be a result of failing to save the fossil
+			// collection file after making it a fossil.
+			if _, found := referencedFossils[file]; !found {
+				if dryRun {
+					LOG_INFO("FOSSIL_UNREFERENCED", "Found unreferenced fossil %s", file)
+					continue
+				}
+
+				chunk := strings.Replace(file, "/", "", -1)
+				chunk = strings.Replace(chunk, ".fsl", "", -1)
+
+				if _, found := referencedChunks[chunk]; found {
+					manager.resurrectChunk(chunkDir+file, chunk)
+				} else {
+					err := manager.storage.DeleteFile(0, chunkDir+file)
+					if err != nil {
+						LOG_WARN("FOSSIL_DELETE", "Failed to remove the unreferenced fossil %s: %v", file, err)
+					} else {
+						LOG_DEBUG("FOSSIL_DELETE", "Deleted unreferenced fossil %s", file)
+						fmt.Fprintf(logFile, "Deleted unreferenced fossil %s\n", file)
+					}
+				}
+			}
+
+			continue
+		}
+
+		chunk := strings.Replace(file, "/", "", -1)
+
+		if !chunkRegex.MatchString(chunk) {
+			LOG_WARN("CHUNK_UNKONWN_FILE", "File %s is not a chunk", file)
+			continue
+		}
+
+		if value, found := referencedChunks[chunk]; !found {
+			if dryRun {
+				LOG_INFO("CHUNK_UNREFERENCED", "Found unreferenced chunk %s", chunk)
+				continue
+			}
+
+			manager.fossilizeChunk(chunk, chunkDir+file, exclusive, collection)
+			if exclusive {
+				fmt.Fprintf(logFile, "Deleted chunk %s (exclusive mode)\n", chunk)
+			} else {
+				fmt.Fprintf(logFile, "Marked fossil %s\n", chunk)
+			}
+		} else if value {
+			// Note that the initial value is false.  So if the value is true it means another copy of the chunk
+			// exists in a higher-level directory.
+
+			if dryRun {
+				LOG_INFO("CHUNK_REDUNDANT", "Found redundant chunk %s", chunk)
+				continue
+			}
+
+			// This is a redundant chunk file (for instance D3/495A8D and D3/49/5A8D )
+			err := manager.storage.DeleteFile(0, chunkDir+file)
+			if err != nil {
+				LOG_WARN("CHUNK_DELETE", "Failed to remove the redundant chunk file %s: %v", file, err)
+			} else {
+				LOG_TRACE("CHUNK_DELETE", "Removed the redundant chunk file %s", file)
+				fmt.Fprintf(logFile, "Deleted redundant chunk %s\n", file)
+			}
+
+		} else {
+			referencedChunks[chunk] = true
+			LOG_DEBUG("CHUNK_KEEP", "Chunk %s is referenced", chunk)
+		}
+	}
+
+	return true
+}
+
 // CheckSnapshot performs sanity checks on the given snapshot.
 func (manager *SnapshotManager) CheckSnapshot(snapshot *Snapshot) (err error) {