Skip to content

New scrubbing feature to validate the database index with the object storage #134

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
TheGuitarleader merged 4 commits into from
Oct 12, 2026
Merged

New scrubbing feature to validate the database index with the object storage #134

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
09f8470
Refactor parallel command handling:
TheGuitarleader Sep 21, 2026
b090765
Streamline `RunCommand` service initialization by removing boolean op…
TheGuitarleader Oct 10, 2026
8f67f22
Improve file scrubbing to handle corrupted files:
TheGuitarleader Oct 10, 2026
e8e4093
Merge branch 'main' into feature/scrub
TheGuitarleader Oct 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
91 changes: 3 additions & 88 deletions Parallel.Cli/Commands/RestoreCommand.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,17 +44,6 @@ public RestoreCommand() : base("restore", "Pulls files a vault.")
this.AddOption(_dryRunOpt);
this.AddOption(_verboseOpt);
this.SetHandler(HandlePullAsync, _sourceOpt, _configOpt, _beforeOpt, _destOpt, _archiveOpt, _forceOpt, _verboseOpt, _dryRunOpt);

this.AddCommand(addCmd);
addCmd.AddArgument(_sourceArg);
addCmd.AddOption(_configOpt);
addCmd.AddOption(_destOpt);
addCmd.SetHandler(HandleAddAsync, _sourceArg, _configOpt, _destOpt);

this.AddCommand(removeCmd);
removeCmd.AddArgument(_sourceArg);
removeCmd.AddOption(_configOpt);
removeCmd.SetHandler(HandleRemoveAsync, _sourceArg, _configOpt);
}

private async Task HandlePullAsync(string? path, string? config, DateTime before, string? destination, bool archive, bool force, bool verbose, bool dryRun)
Expand Down Expand Up @@ -95,10 +84,10 @@ private async Task PullSystemAsync(LocalVaultConfig vault, DateTime timestamp, s
return;
}

foreach (PullRecord record in syncManager.RemoteVault.PullDirectories.Where(r => r.Machine == Environment.MachineName))
/*foreach (PullRecord record in syncManager.RemoteVault.PullDirectories.Where(r => r.Machine == Environment.MachineName))
{
await PullInternalAsync(syncManager, record.Source, timestamp, (output ?? record.Destination), archive, force, verbose, dryRun);
}
}*/
}

private async Task PullPathAsync(LocalVaultConfig vault, string path, DateTime timestamp, string? destination, bool archive, bool force, bool verbose, bool dryRun)
Expand Down Expand Up @@ -148,85 +137,11 @@ private async Task PullInternalAsync(ISyncManager syncManager, string path, Date
{
CommandLine.WriteLine(syncManager.RemoteVault, $"Pulling {restoreFiles.Count:N0} files...", ConsoleColor.DarkGray);
IProgressReporter progressReporter = verbose ? new ProgressReporter(syncManager.RemoteVault, restoreFiles.Count) : new LoggingProgressReporter(syncManager.RemoteVault);
int pulledFiles = await syncManager.PullFilesAsync(restoreFiles.ToArray(), progressReporter);
int pulledFiles = await syncManager.RestoreFilesAsync(restoreFiles.ToArray(), progressReporter);

CommandLine.WriteLine(syncManager.RemoteVault, $"Successfully pulled {pulledFiles:N0} files in {_sw.Elapsed}.", ConsoleColor.Green);
await syncManager.DisconnectAsync();
}
}

#region Add

private async Task HandleAddAsync(string path, string? config, string? destination)
{
LocalVaultConfig? localVault = string.IsNullOrEmpty(config) ? ParallelConfig.Load().Vaults.FirstOrDefault(v => v.Enabled) : ParallelConfig.GetVault(config);
if (!string.IsNullOrEmpty(config) && localVault != null)
{
await AddPathAsync(localVault, path, destination);
}
else
{
await Program.Settings.ForEachVaultAsync(vault => AddPathAsync(vault, path, destination));
}
}

private async Task AddPathAsync(LocalVaultConfig vault, string path, string? destination)
{
ISyncManager? syncManager = SyncManager.CreateNew(vault);
if (syncManager == null || !await syncManager.ConnectAsync())
{
CommandLine.WriteLine(vault, "Failed to connect to vault!", ConsoleColor.Red);
return;
}

PullRecord record = new(path, destination);
if (!syncManager.RemoteVault.PullDirectories.Add(record))
{
CommandLine.WriteLine(vault, $"Unable to add path: '{path}'", ConsoleColor.Yellow);
return;
}

CommandLine.WriteLine(vault, $"Successfully added '{path}'", ConsoleColor.Green);
await syncManager.DisconnectAsync();
}

#endregion

#region Remove

private async Task HandleRemoveAsync(string path, string? config)
{
LocalVaultConfig? localVault = string.IsNullOrEmpty(config) ? ParallelConfig.Load().Vaults.FirstOrDefault(v => v.Enabled) : ParallelConfig.GetVault(config);
if (!string.IsNullOrEmpty(config) && localVault != null)
{
await RemovePathAsync(localVault, path);
}
else
{
await Program.Settings.ForEachVaultAsync(vault => RemovePathAsync(vault, path));
}
}

private async Task RemovePathAsync(LocalVaultConfig vault, string path)
{
ISyncManager? syncManager = SyncManager.CreateNew(vault);
if (syncManager == null || !await syncManager.ConnectAsync())
{
CommandLine.WriteLine(vault, "Failed to connect to vault!", ConsoleColor.Red);
return;
}

IEnumerable<PullRecord> records = syncManager.RemoteVault.PullDirectories.Where(r => r.Machine == Environment.MachineName && r.Source == path);
foreach (PullRecord record in records)
{
if (syncManager.RemoteVault.PullDirectories.Remove(record)) continue;
CommandLine.WriteLine(vault, $"Unable to remove path: '{path}'", ConsoleColor.Yellow);
}

CommandLine.WriteLine(vault, $"Successfully removed '{path}'", ConsoleColor.Green);
await syncManager.DisconnectAsync();
}

#endregion
}
}
17 changes: 13 additions & 4 deletions Parallel.Cli/Commands/ScrubCommand.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,7 +64,7 @@ private async Task ScrubSystemAsync(LocalVaultConfig vault, bool verbose)
return;
}

foreach (string path in syncManager.RemoteVault.PushDirectories)
foreach (string path in syncManager.RemoteVault.SyncDirectories)
{
await ScrubInternalAsync(syncManager, path, verbose);
}
Expand All @@ -88,15 +88,24 @@ private async Task ScrubInternalAsync(ISyncManager syncManager, string path, boo
IReadOnlyList<LocalFile> files = await (syncManager.Database?.GetFilesAsync(path, DateTime.Now) ?? Task.FromResult<IReadOnlyList<LocalFile>>([]));
if (files.Count == 0)
{
CommandLine.WriteLine($"No prunable files were found!", ConsoleColor.Yellow);
CommandLine.WriteLine($"No files were found!", ConsoleColor.Yellow);
return;
}

CommandLine.WriteLine(syncManager.RemoteVault, $"Scrubbing {files.Count:N0} files...", ConsoleColor.DarkGray);
IProgressReporter progressReporter = verbose ? new ProgressReporter(syncManager.RemoteVault, files.Count) : new LoggingProgressReporter(syncManager.RemoteVault);
int scrubbedFiles = await syncManager.ScrubFilesAsync(files, progressReporter);
int validFiles = await syncManager.ScrubFilesAsync(files, progressReporter);
if (validFiles == files.Count)
{
CommandLine.WriteLine(syncManager.RemoteVault, $"Successfully scrubbed {files.Count:N0} files in {_sw.Elapsed}.", ConsoleColor.Green);
}
else
{
int corrupted = files.Count - validFiles;
CommandLine.WriteLine(syncManager.RemoteVault, $"Removed {corrupted:N0} corrupted or missing files in {_sw.Elapsed}.", ConsoleColor.Yellow);
}

CommandLine.WriteLine(syncManager.RemoteVault, $"Successfully scrubbed {scrubbedFiles:N0} files in {_sw.Elapsed}.", ConsoleColor.Green);

await syncManager.DisconnectAsync();
}
}
Expand Down
2 changes: 1 addition & 1 deletion Parallel.Cli/Commands/SnapshotsCommand.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,7 +109,7 @@ private async Task CreateSnapshotAsync(LocalVaultConfig vault)
}

ConcurrentBag<SnapshotRecord> snapshots = new();
await System.Threading.Tasks.Parallel.ForEachAsync(syncManager.RemoteVault.PushDirectories, ParallelConfig.Options, async (path, ct) =>
await System.Threading.Tasks.Parallel.ForEachAsync(syncManager.RemoteVault.SyncDirectories, ParallelConfig.Options, async (path, ct) =>
{
IReadOnlyList<LocalFile> files = await (syncManager.Database?.GetLatestFilesAsync(path, DateTime.UtcNow, false) ?? Task.FromResult<IReadOnlyList<LocalFile>>([]));
foreach (LocalFile file in files) snapshots.Add(new SnapshotRecord(file));
Expand Down
54 changes: 33 additions & 21 deletions Parallel.Cli/Commands/SyncCommand.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
using Parallel.Core.IO.Syncing;
using Parallel.Core.Models;
using Parallel.Core.Settings;
using Parallel.Core.Utils;

namespace Parallel.Cli.Commands
{
Expand All @@ -33,7 +34,8 @@ public SyncCommand() : base("sync", "Syncs system files with vaults.")
this.AddOption(_configOpt);
this.AddOption(_forceOpt);
this.AddOption(_verboseOpt);
this.SetHandler(HandlePushAsync, _sourceOpt, _configOpt, _forceOpt, _verboseOpt);
this.AddOption(_dryRunOpt);
this.SetHandler(HandleSyncAsync, _sourceOpt, _configOpt, _forceOpt, _verboseOpt, _dryRunOpt);

this.AddCommand(addCmd);
addCmd.AddArgument(_sourceArg);
Expand All @@ -46,37 +48,37 @@ public SyncCommand() : base("sync", "Syncs system files with vaults.")
removeCmd.SetHandler(HandleRemoveAsync, _sourceArg, _configOpt);
}

#region Push
#region Sync

private async Task HandlePushAsync(string? path, string? config, bool force, bool verbose)
private async Task HandleSyncAsync(string? path, string? config, bool force, bool verbose, bool dryRun)
{
_sw = Stopwatch.StartNew();
LocalVaultConfig? localVault = ParallelConfig.GetVault(config);
if (localVault != null)
{
if (!string.IsNullOrEmpty(path))
{
await PushPathAsync(localVault, path, force, verbose);
await SyncPathAsync(localVault, path, force, verbose, dryRun);
}
else
{
await PushSystemAsync(localVault, force, verbose);
await SyncSystemAsync(localVault, force, verbose, dryRun);
}
}
else
{
if (!string.IsNullOrEmpty(path))
{
await Program.Settings.ForEachVaultAsync(vault => PushPathAsync(vault, path, force, verbose));
await Program.Settings.ForEachVaultAsync(vault => SyncPathAsync(vault, path, force, verbose, dryRun));
}
else
{
await Program.Settings.ForEachVaultAsync(vault => PushSystemAsync(vault, force, verbose));
await Program.Settings.ForEachVaultAsync(vault => SyncSystemAsync(vault, force, verbose, dryRun));
}
}
}

private async Task PushSystemAsync(LocalVaultConfig vault, bool force, bool verbose)
private async Task SyncSystemAsync(LocalVaultConfig vault, bool force, bool verbose, bool dryRun)
{
ISyncManager? syncManager = SyncManager.CreateNew(vault);
if (syncManager == null || !await syncManager.ConnectAsync())
Expand All @@ -85,15 +87,15 @@ private async Task PushSystemAsync(LocalVaultConfig vault, bool force, bool verb
return;
}

foreach (string path in syncManager.RemoteVault.PushDirectories)
foreach (string path in syncManager.RemoteVault.SyncDirectories)
{
await PushInternalAsync(syncManager, path, force, verbose);
await SyncInternalAsync(syncManager, path, force, verbose, dryRun);
}

await syncManager.DisconnectAsync();
}

private async Task PushPathAsync(LocalVaultConfig vault, string path, bool force, bool verbose)
private async Task SyncPathAsync(LocalVaultConfig vault, string path, bool force, bool verbose, bool dryRun)
{
ISyncManager? syncManager = SyncManager.CreateNew(vault);
if (syncManager == null || !await syncManager.ConnectAsync())
Expand All @@ -102,14 +104,14 @@ private async Task PushPathAsync(LocalVaultConfig vault, string path, bool force
return;
}

await PushInternalAsync(syncManager, path, force, verbose);
await SyncInternalAsync(syncManager, path, force, verbose, dryRun);
await syncManager.DisconnectAsync();
}

private async Task PushInternalAsync(ISyncManager syncManager, string path, bool force, bool verbose)
private async Task SyncInternalAsync(ISyncManager syncManager, string path, bool force, bool verbose, bool dryRun)
{
// Normalize paths for safe comparison
string[] backupFolders = syncManager.RemoteVault.PushDirectories.ToArray();
string[] backupFolders = syncManager.RemoteVault.SyncDirectories.ToArray();
string[] ignoredFolders = syncManager.RemoteVault.IgnoreDirectories.ToArray();

bool isFile = PathBuilder.IsFile(path);
Expand All @@ -134,12 +136,22 @@ private async Task PushInternalAsync(ISyncManager syncManager, string path, bool
CommandLine.WriteLine(syncManager.RemoteVault, $"The provided {(isFile ? "file" : "folder")} is already up to date.", ConsoleColor.Green);
return;
}

if (dryRun)
{
string fileName = PathBuilder.TempFile;
await File.WriteAllLinesAsync(fileName, files.Select(f => f.Fullname).OrderBy(f => f));
CommandLine.WriteLine($"This operation will sync {files.Length:N0} files ({Formatter.FromBytes(files.Sum(f => f.LocalSize))})", ConsoleColor.Green);
CommandLine.WriteLine($"A detailed list can be found here: {fileName}", ConsoleColor.DarkGray);
}
else
{
CommandLine.WriteLine(syncManager.RemoteVault, $"Syncing {files.Length:N0} files...", ConsoleColor.DarkGray);
IProgressReporter progressReporter = verbose ? new ProgressReporter(syncManager.RemoteVault, successFiles) : new LoggingProgressReporter(syncManager.RemoteVault);
int pushedFiles = await syncManager.SyncFilesAsync(files, progressReporter, force);

CommandLine.WriteLine(syncManager.RemoteVault, $"Pushing {files.Length:N0} files...", ConsoleColor.DarkGray);
IProgressReporter progressReporter = verbose ? new ProgressReporter(syncManager.RemoteVault, successFiles) : new LoggingProgressReporter(syncManager.RemoteVault);
int pushedFiles = await syncManager.PushFilesAsync(files, progressReporter, force);

CommandLine.WriteLine(syncManager.RemoteVault, $"Successfully pushed {pushedFiles:N0} files in {_sw.Elapsed}.", ConsoleColor.Green);
CommandLine.WriteLine(syncManager.RemoteVault, $"Successfully pushed {pushedFiles:N0} files in {_sw.Elapsed}.", ConsoleColor.Green);
}
}

#endregion
Expand Down Expand Up @@ -168,7 +180,7 @@ private async Task AddPathAsync(LocalVaultConfig vault, string path)
return;
}

if (!syncManager.RemoteVault.PushDirectories.Add(path))
if (!syncManager.RemoteVault.SyncDirectories.Add(path))
{
CommandLine.WriteLine(vault, $"Unable to add path: '{path}'", ConsoleColor.Yellow);
return;
Expand Down Expand Up @@ -204,7 +216,7 @@ private async Task RemovePathAsync(LocalVaultConfig vault, string path)
return;
}

if (!syncManager.RemoteVault.PushDirectories.Remove(path))
if (!syncManager.RemoteVault.SyncDirectories.Remove(path))
{
CommandLine.WriteLine(vault, $"Unable to remove path: '{path}'", ConsoleColor.Yellow);
return;
Expand Down
5 changes: 2 additions & 3 deletions Parallel.Cli/Commands/VaultsCommand.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,7 @@ public VaultsCommand() : base("vaults", "View or edit the vaults.")
string bucketName = string.IsNullOrEmpty(bucketInput) ? "parallel" : bucketInput;
spc.RootDirectory = bucketName;

string? regionInput = CommandLine.ReadString("Region Name (Leave empty for default)");
string? regionInput = CommandLine.ReadString("Region (Leave empty for default)");
string regionName = string.IsNullOrEmpty(regionInput) ? "us-east-1" : regionInput;
spc.Region = regionName;

Expand Down Expand Up @@ -113,8 +113,7 @@ public VaultsCommand() : base("vaults", "View or edit the vaults.")

RemoteVaultConfig remoteVault = syncManager.RemoteVault;
CommandLine.WriteLine($"'{remoteVault.Name}' ({remoteVault.Id}):");
CommandLine.WriteArray("Push Directories", remoteVault.PushDirectories);
CommandLine.WriteArray("Pull Directories", remoteVault.PullDirectories.Select(d => d.Source));
CommandLine.WriteArray("Push Directories", remoteVault.SyncDirectories);
CommandLine.WriteArray("Ignore Directories", remoteVault.IgnoreDirectories);
CommandLine.WriteArray("Prune Directories", remoteVault.PruneDirectories);
CommandLine.WriteLine($"Prune Period: {remoteVault.PrunePeriod} days");
Expand Down
18 changes: 10 additions & 8 deletions Parallel.Cli/Services/RunCommand.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Parallel.Core.Tasks;

namespace Parallel.Cli.Services
{
Expand All @@ -15,22 +16,23 @@ public class RunCommand : Command

public RunCommand() : base("run", "Starts the background services.")
{
this.AddOption(_cleanOpt);
this.AddOption(_syncOpt);
this.AddOption(_pruneOpt);
this.SetHandler(HandleCommandAsync, _cleanOpt, _pruneOpt, _syncOpt);
this.SetHandler(HandleCommandAsync);
}

private async Task HandleCommandAsync(bool clean, bool prune, bool sync)
private async Task HandleCommandAsync()
{
HostApplicationBuilder builder = Host.CreateApplicationBuilder();
if (OperatingSystem.IsWindows()) builder.Services.AddWindowsService();
if (OperatingSystem.IsLinux()) builder.Services.AddSystemd();

if (clean) builder.Services.AddHostedService<CleanService>();
if (prune) builder.Services.AddHostedService<PruneService>();
if (sync) builder.Services.AddHostedService<SyncService>();
builder.Services.AddHostedService<LifetimeService>();
builder.Services.AddHostedService<CleanService>();
builder.Services.AddHostedService<PruneService>();
builder.Services.AddHostedService<SyncService>();

builder.Services.AddHostedService<TaskService>();
builder.Services.AddSingleton<TaskQueueExecutor>();
builder.Services.AddSingleton<TaskQueuer>();

builder.Logging.ClearProviders();
builder.Logging.AddSerilog();
Expand Down
Loading
Loading