Fixed incorrect stats during backup; also check in files missing from last commit

duplicacy/duplicacy_main.go

@@ -1683,7 +1683,7 @@ func main() {
     app.Name = "duplicacy"
     app.HelpName = "duplicacy"
     app.Usage = "A new generation cloud backup tool based on lock-free deduplication"
-    app.Version = "2.0.4"
+    app.Version = "2.0.5"
 
     // If the program is interrupted, call the RunAtError function.
     c := make(chan os.Signal, 1)                                       

integration_tests/resume_test.sh

@@ -6,32 +6,25 @@
 fixture
 
 pushd ${TEST_REPO}
-${DUPLICACY} init integration-tests $TEST_STORAGE -c 4k
+${DUPLICACY} init integration-tests $TEST_STORAGE -c  4 
 
-# Create 10 20k files
-add_file file1 20000
-add_file file2 20000
-add_file file3 20000
-add_file file4 20000
-add_file file5 20000
-add_file file6 20000
-add_file file7 20000
-add_file file8 20000
-add_file file9 20000
-add_file file10 20000
+# Create 10 20 files
+add_file file1 20
+add_file file2 20
+add_file file3 20
+add_file file4 20
+add_file file5 20
+add_file file6 20
+add_file file7 20
+add_file file8 20
+add_file file9 20
+add_file file10 20
 
-# Limit the rate to 10k/s so the backup will take about 10 seconds
-${DUPLICACY} backup -limit-rate 10 -threads 4 &
-# Kill the backup after 3 seconds
-DUPLICACY_PID=$!
-sleep 3
-kill -2 ${DUPLICACY_PID}
+# Fail at the 10th chunk
+env DUPLICACY_FAIL_CHUNK=10 ${DUPLICACY} backup
 
 # Try it again to test the multiple-resume case
-${DUPLICACY} backup -limit-rate 10  -threads 4&
-DUPLICACY_PID=$!
-sleep 3
-kill -2 ${DUPLICACY_PID}
+env DUPLICACY_FAIL_CHUNK=5 ${DUPLICACY} backup
 
 # Fail the backup before uploading the snapshot
 env DUPLICACY_FAIL_SNAPSHOT=true ${DUPLICACY} backup

src/duplicacy_b2client.go

@@ -135,7 +135,7 @@ func (client *B2Client) call(url string, input interface{}) (io.ReadCloser, int6
             return nil, 0, err
         }
 
-        if response.StatusCode < 400 {
+        if response.StatusCode < 300 {
             return response.Body, response.ContentLength, nil
         }
 
@@ -160,6 +160,10 @@ func (client *B2Client) call(url string, input interface{}) (io.ReadCloser, int6
         } else if response.StatusCode >= 500 && response.StatusCode <= 599 {
             backoff = client.retry(backoff, response)
             continue
+        } else {
+            LOG_INFO("BACKBLAZE_CALL", "URL request '%s' returned status code %d", url, response.StatusCode)
+            backoff = client.retry(backoff, response)
+            continue
         }
 
         defer response.Body.Close()
@@ -487,7 +491,7 @@ func (client *B2Client) UploadFile(filePath string, content []byte, rateLimit in
         io.Copy(ioutil.Discard, response.Body)
         response.Body.Close()
 
-        if response.StatusCode < 400 {
+        if response.StatusCode < 300 {
             return nil
         }
 

src/duplicacy_backupmanager.go

@@ -142,6 +142,12 @@ func setEntryContent(entries[] *Entry, chunkLengths[]int, offset int) {
         }
         totalChunkSize += int64(length)
     }
+
+    // If there are some unvisited entries (which happens when saving an incomplete snapshot), 
+    // set their sizes to -1 so they won't be saved to the incomplete snapshot
+    for j := i; j < len(entries); j++ {
+        entries[j].Size = -1
+    }
 }
 
 // Backup creates a snapshot for the repository 'top'.  If 'quickMode' is true, only files with different sizes
@@ -248,15 +254,15 @@ func (manager *BackupManager) Backup(top string, quickMode bool, threads int, ta
             incompleteSnapshot.ChunkHashes = incompleteSnapshot.ChunkHashes[:lastCompleteChunk + 1]
             incompleteSnapshot.ChunkLengths = incompleteSnapshot.ChunkLengths[:lastCompleteChunk + 1]            
             remoteSnapshot = incompleteSnapshot
+            LOG_INFO("FILE_SKIP", "Skipped %d files from previous incomplete backup", len(files))
         }
 
     }
 
-    var numberOfNewFileChunks int                 // number of new file chunks
+    var numberOfNewFileChunks int64                 // number of new file chunks
     var totalUploadedFileChunkLength int64        // total length of uploaded file chunks
     var totalUploadedFileChunkBytes int64         // how many actual bytes have been uploaded
 
-    var numberOfNewSnapshotChunks int             // number of new snapshot chunks
     var totalUploadedSnapshotChunkLength int64    // size of uploaded snapshot chunks
     var totalUploadedSnapshotChunkBytes int64     // how many actual bytes have been uploaded
 
@@ -358,6 +364,10 @@ func (manager *BackupManager) Backup(top string, quickMode bool, threads int, ta
     // the file reader implements the Reader interface. When an EOF is encounter, it opens the next file unless it
     // is the last file.
     fileReader := CreateFileReader(shadowTop, modifiedEntries)
+    // Set all file sizes to -1 to indicate they haven't been processed
+    for _, entry := range modifiedEntries {
+        entry.Size = -1
+    }
 
     startUploadingTime := time.Now().Unix()
 
@@ -374,6 +384,14 @@ func (manager *BackupManager) Backup(top string, quickMode bool, threads int, ta
         keepUploadAlive = int64(value)
     }
 
+    // Fail at the chunk specified by DUPLICACY_FAIL_CHUNK to simulate a backup error
+    chunkToFail := -1
+    if value, found := os.LookupEnv("DUPLICACY_FAIL_CHUNK"); found {
+        chunkToFail, _ = strconv.Atoi(value)
+        LOG_INFO("SNAPSHOT_FAIL", "Will abort the backup on chunk %d", chunkToFail)
+    }
+
+
     chunkMaker := CreateChunkMaker(manager.config, false)
     chunkUploader := CreateChunkUploader(manager.config, manager.storage, nil, threads, nil)
 
@@ -388,16 +406,16 @@ func (manager *BackupManager) Backup(top string, quickMode bool, threads int, ta
                     if !localSnapshotReady {
                         // Lock it to gain exclusive access to uploadedChunkHashes and uploadedChunkLengths
                         uploadedChunkLock.Lock()
-                        for _, entry := range uploadedEntries {
-                            entry.EndChunk = -1
-                        }
                         setEntryContent(uploadedEntries, uploadedChunkLengths, len(preservedChunkHashes))
                         if len(preservedChunkHashes) > 0 {
+                            //localSnapshot.Files = preservedEntries
+                            //localSnapshot.Files = append(preservedEntries, uploadedEntries...)
                             localSnapshot.ChunkHashes = preservedChunkHashes
                             localSnapshot.ChunkHashes = append(localSnapshot.ChunkHashes, uploadedChunkHashes...)
                             localSnapshot.ChunkLengths = preservedChunkLengths
                             localSnapshot.ChunkLengths = append(localSnapshot.ChunkLengths, uploadedChunkLengths...)
                         } else {
+                            //localSnapshot.Files = uploadedEntries
                             localSnapshot.ChunkHashes = uploadedChunkHashes
                             localSnapshot.ChunkLengths = uploadedChunkLengths
                         }
@@ -428,16 +446,16 @@ func (manager *BackupManager) Backup(top string, quickMode bool, threads int, ta
                 LOG_DEBUG("CHUNK_CACHE", "Skipped chunk %s in cache", chunk.GetID())
             } else {
                 if uploadSize > 0 {
-                    numberOfNewFileChunks++
-                    totalUploadedFileChunkLength += int64(chunkSize)
-                    totalUploadedFileChunkBytes += int64(uploadSize)
+                    atomic.AddInt64(&numberOfNewFileChunks, 1)
+                    atomic.AddInt64(&totalUploadedFileChunkLength, int64(chunkSize))
+                    atomic.AddInt64(&totalUploadedFileChunkBytes, int64(uploadSize))
                     action = "Uploaded"
                 } else {
                     LOG_DEBUG("CHUNK_EXIST", "Skipped chunk %s in the storage", chunk.GetID())
                 }
             }
 
-            uploadedModifiedFileSize += int64(chunkSize)
+            uploadedModifiedFileSize := atomic.AddInt64(&uploadedModifiedFileSize, int64(chunkSize))
 
             if IsTracing() || showStatistics {
                 now := time.Now().Unix()
@@ -494,21 +512,27 @@ func (manager *BackupManager) Backup(top string, quickMode bool, threads int, ta
                 uploadedChunkLengths = append(uploadedChunkLengths, chunkSize)
                 uploadedChunkLock.Unlock()
 
+                if len(uploadedChunkHashes) == chunkToFail {
+                    LOG_ERROR("SNAPSHOT_FAIL", "Artificially fail the chunk %d for testing purposes", chunkToFail)
+                }
+
             },
             func (fileSize int64, hash string) (io.Reader, bool) {
+
+                // Must lock here because the RunAtError function called by other threads may access uploadedEntries
+                uploadedChunkLock.Lock()
+                defer uploadedChunkLock.Unlock()
+
                 // This function is called when a new file is needed
                 entry := fileReader.CurrentEntry
                 entry.Hash = hash
-                if entry.Size != fileSize {
-                    totalModifiedFileSize += fileSize - entry.Size
-                    entry.Size = fileSize
-                }
+                entry.Size = fileSize
                 uploadedEntries = append(uploadedEntries, entry)
 
                 if !showStatistics || IsTracing() || RunInBackground {
                     LOG_INFO("PACK_END", "Packed %s (%d)", entry.Path, entry.Size)
                 }
-
+                
                 fileReader.NextFile()
 
                 if fileReader.CurrentFile != nil {
@@ -544,6 +568,7 @@ func (manager *BackupManager) Backup(top string, quickMode bool, threads int, ta
 
     err = manager.SnapshotManager.CheckSnapshot(localSnapshot)
     if err != nil {
+        RunAtError = func() {} // Don't save the incomplete snapshot
         LOG_ERROR("SNAPSHOT_CHECK", "The snapshot contains an error: %v", err)
         return false
     }
@@ -630,7 +655,7 @@ func (manager *BackupManager) Backup(top string, quickMode bool, threads int, ta
         LOG_INFO("BACKUP_STATS", "All chunks: %d total, %s bytes; %d new, %s bytes, %s bytes uploaded",
                 len(localSnapshot.ChunkHashes) + totalSnapshotChunks,
                 PrettyNumber(totalFileChunkLength + totalSnapshotChunkLength),
-                numberOfNewFileChunks + numberOfNewSnapshotChunks,
+                int(numberOfNewFileChunks) + numberOfNewSnapshotChunks,
                 PrettyNumber(totalUploadedFileChunkLength + totalUploadedSnapshotChunkLength),
                 PrettyNumber(totalUploadedFileChunkBytes + totalUploadedSnapshotChunkBytes))
 
@@ -749,6 +774,7 @@ func (manager *BackupManager) Restore(top string, revision int, inPlace bool, qu
     i := 0
     for _, entry := range remoteSnapshot.Files {
 
+        skipped := false
         // Find local files that don't exist in the remote snapshot
         for i < len(localSnapshot.Files) {
             local := localSnapshot.Files[i]
@@ -760,11 +786,18 @@ func (manager *BackupManager) Restore(top string, revision int, inPlace bool, qu
             } else {
                 if compare == 0 {
                     i++
+                    if quickMode && local.IsSameAs(entry) {
+                        skipped = true
+                    }
                 }
                 break
             }
         }
 
+        if skipped {
+            continue
+        }
+
         fullPath := joinPath(top, entry.Path)
         if entry.IsLink() {
             stat, err := os.Lstat(fullPath)

src/duplicacy_gcdstorage.go

@@ -32,6 +32,7 @@ type GCDStorage struct {
     idCacheLock *sync.Mutex
     backoff int
 
+    isConnected bool
     numberOfThreads int
     TestMode bool
 
@@ -64,6 +65,12 @@ func (storage *GCDStorage) shouldRetry(err error) (bool, error) {
             // User Rate Limit Exceeded
             message = "User Rate Limit Exceeded"
             retry = true
+        } else if e.Code == 401 {
+            // Only retry on authorization error when storage has been connected before
+            if storage.isConnected {
+                message = "Authorization Error"
+                retry = true
+            }
         }
     } else if e, ok := err.(*url.Error); ok {
         message = e.Error()
@@ -295,6 +302,8 @@ func CreateGCDStorage(tokenFile string, storagePath string, threads int) (storag
         }
     }
 
+    storage.isConnected = true
+
     return storage, nil
 
 }

src/duplicacy_snapshot.go

@@ -149,6 +149,7 @@ func LoadIncompleteSnapshot() (snapshot *Snapshot) {
     snapshotFile := path.Join(GetDuplicacyPreferencePath(), "incomplete")
     description, err := ioutil.ReadFile(snapshotFile)    
     if err != nil {
+        LOG_DEBUG("INCOMPLETE_LOCATE", "Failed to locate incomplete snapshot: %v", err)
         return nil
     }
 
@@ -156,6 +157,7 @@ func LoadIncompleteSnapshot() (snapshot *Snapshot) {
 
     err = json.Unmarshal(description, &incompleteSnapshot)
     if err != nil {
+        LOG_DEBUG("INCOMPLETE_PARSE", "Failed to parse incomplete snapshot: %v", err)
         return nil
     }
 
@@ -163,6 +165,7 @@ func LoadIncompleteSnapshot() (snapshot *Snapshot) {
     for _, chunkHash := range incompleteSnapshot.ChunkHashes {
         hash, err := hex.DecodeString(chunkHash)
         if err != nil {
+            LOG_DEBUG("INCOMPLETE_DECODE", "Failed to decode incomplete snapshot: %v", err)
             return nil
         }
         chunkHashes = append(chunkHashes, string(hash))
@@ -181,7 +184,8 @@ func LoadIncompleteSnapshot() (snapshot *Snapshot) {
 func SaveIncompleteSnapshot(snapshot *Snapshot) {
     var files []*Entry
     for _, file := range snapshot.Files {
-        if file.EndChunk >= 0 {
+        // All unprocessed files will have a size of -1
+        if file.Size >= 0 {
             file.Attributes = nil
             files = append(files, file)
         } else {
@@ -199,7 +203,7 @@ func SaveIncompleteSnapshot(snapshot *Snapshot) {
         ChunkLengths: snapshot.ChunkLengths,
     }
 
-    description, err := json.Marshal(incompleteSnapshot)
+    description, err := json.MarshalIndent(incompleteSnapshot, "", "  ")
     if err != nil {
         LOG_WARN("INCOMPLETE_ENCODE", "Failed to encode the incomplete snapshot: %v", err)
         return

src/duplicacy_snapshotmanager.go

@@ -2181,7 +2181,9 @@ func (manager *SnapshotManager) CheckSnapshot(snapshot *Snapshot) (err error) {
     if len(entries) > 0 && entries[0].StartChunk != 0 {
         return fmt.Errorf("The first file starts at chunk %d", entries[0].StartChunk )
     }
-    if lastChunk < numberOfChunks - 1 {
+
+    // There may be a last chunk whose size is 0 so we allow this to happen
+    if lastChunk < numberOfChunks - 2 {
         return fmt.Errorf("The last file ends at chunk %d but the number of chunks is %d", lastChunk, numberOfChunks)
     }