feat(rst): cleanup incomplete downloads

Author: swartzn

common/rst/rst.go

@@ -327,8 +327,8 @@ func IsFileLocked(lockedInfo *flex.JobLockedInfo) bool {
 	return lockedInfo != nil && lockedInfo.ReadWriteLocked
 }
 
-// FileExists returns whether the file exists.
-func FileExists(lockedInfo *flex.JobLockedInfo) bool {
+// FileExistedBeforeJob returns whether the file exists when the job was created.
+func FileExistedBeforeJob(lockedInfo *flex.JobLockedInfo) bool {
 	return lockedInfo.Exists
 }
 
@@ -378,7 +378,7 @@ func BuildJobRequest(ctx context.Context, client Provider, mountPoint filesystem
 	}
 
 	lockedInfo := cfg.LockedInfo
-	if !IsFileLocked(lockedInfo) && FileExists(lockedInfo) {
+	if !IsFileLocked(lockedInfo) && FileExistedBeforeJob(lockedInfo) {
 		return getRequestWithFailedPrecondition("path lock has not been acquired")
 
 	}
@@ -397,7 +397,7 @@ func BuildJobRequest(ctx context.Context, client Provider, mountPoint filesystem
 	}
 
 	if cfg.Download && cfg.RemotePath == "" {
-		if !FileExists(lockedInfo) {
+		if !FileExistedBeforeJob(lockedInfo) {
 			return getRequestWithFailedPrecondition(fmt.Sprintf("unable to determine remote path: %s", fs.ErrNotExist.Error()))
 		}
 
@@ -464,7 +464,7 @@ func PrepareFileStateForWorkRequests(ctx context.Context, client Provider, mount
 
 	alreadySynced := IsFileAlreadySynced(lockedInfo)
 	if cfg.StubLocal {
-		if (cfg.Download && (cfg.Overwrite || !FileExists(lockedInfo))) || alreadySynced {
+		if (cfg.Download && (cfg.Overwrite || !FileExistedBeforeJob(lockedInfo))) || alreadySynced {
 			if mappings == nil {
 				mappings, err = util.GetMappings(ctx)
 				if err != nil && !errors.Is(err, util.ErrMappingRSTs) {
@@ -492,11 +492,11 @@ func PrepareFileStateForWorkRequests(ctx context.Context, client Provider, mount
 			return ErrJobAlreadyOffloaded
 		}
 
-		if cfg.Download && !cfg.Overwrite && FileExists(lockedInfo) {
+		if cfg.Download && !cfg.Overwrite && FileExistedBeforeJob(lockedInfo) {
 			err = fmt.Errorf("download would overwrite existing path but the overwrite flag was not set: %w", fs.ErrExist)
 			return
 		}
-	} else if FileExists(lockedInfo) {
+	} else if FileExistedBeforeJob(lockedInfo) {
 		if alreadySynced {
 			return GetErrJobAlreadyCompleteWithMtime(lockedInfo.Mtime.AsTime())
 		}
@@ -552,8 +552,8 @@ func PrepareFileStateForWorkRequests(ctx context.Context, client Provider, mount
 			err = fmt.Errorf("failed to collect information for new file: %w", err)
 			return
 		}
+		lockedInfo.SetExists(false) // Setting to false since the file did not previously exist.
 		lockedInfo.SetReadWriteLocked(info.ReadWriteLocked)
-		lockedInfo.SetExists(info.Exists)
 		lockedInfo.SetSize(info.Size)
 		lockedInfo.SetMtime(info.Mtime)
 		lockedInfo.SetMode(info.Mode)
@@ -592,7 +592,7 @@ func GetLockedInfo(ctx context.Context, mountPoint filesystem.Provider, mappings
 		}
 		return
 	}
-	lockedInfo.Exists = true
+	lockedInfo.SetExists(true)
 
 	if rstIds == nil {
 		rstIds = entryInfo.Entry.Remote.RSTIDs

common/rst/s3.go

@@ -424,9 +424,27 @@ func (r *S3Client) completeSyncWorkRequests_Download(ctx context.Context, job *b
 	}
 	job.SetStopMtime(timestamppb.New(mtime))
 
-	// Skip checking the file was modified if we were told to abort since the mtime may not have
-	// been set correctly anyway given the error check is skipped above.
-	if !abort {
+	if abort {
+		// If the file previously existed, it is converted into a stub file. Otherwise, the file is
+		// deleted.
+		lockedInfo := sync.LockedInfo
+		if FileExistedBeforeJob(lockedInfo) {
+			mappings, err := util.GetMappings(ctx)
+			if err != nil {
+				return fmt.Errorf("failed to leave stub file: %w", err)
+			}
+
+			err = CreateOffloadedDataFile(ctx, r.mountPoint, mappings, request.Path, sync.RemotePath, request.RemoteStorageTarget, true)
+			if err != nil {
+				return fmt.Errorf("failed to leave stub file: %w", err)
+			}
+			job.GetStatus().SetState(beeremote.Job_OFFLOADED)
+		} else {
+			r.mountPoint.Remove(request.Path)
+		}
+	} else {
+		// Skip checking the file was modified if we were told to abort since the mtime may not have
+		// been set correctly anyway given the error check is skipped above.
 		start := job.GetStartMtime().AsTime()
 		stop := job.GetStopMtime().AsTime()
 		if !start.Equal(stop) {