Add Explorer Workbench with hierarchical, off-UI Storage analysis.
Storage queries run in the background with cancellation and covering indexes so switching views no longer freezes the UI. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
108
src/Explorer.Indexing/DirectoryWatcherHub.cs
Normal file
108
src/Explorer.Indexing/DirectoryWatcherHub.cs
Normal file
@@ -0,0 +1,108 @@
|
||||
using Explorer.Domain;
|
||||
using Explorer.Domain.Abstractions;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Explorer.Indexing;
|
||||
|
||||
public sealed class DirectoryWatcherHub : IDisposable
|
||||
{
|
||||
private readonly IVolumeService _volumes;
|
||||
private readonly IndexingCoordinator _indexing;
|
||||
private readonly IIndexStore _store;
|
||||
private readonly ILogger<DirectoryWatcherHub> _logger;
|
||||
private readonly Dictionary<long, FileSystemWatcher> _watchers = new();
|
||||
private readonly object _gate = new();
|
||||
|
||||
public DirectoryWatcherHub(
|
||||
IVolumeService volumes,
|
||||
IndexingCoordinator indexing,
|
||||
IIndexStore store,
|
||||
ILogger<DirectoryWatcherHub> logger)
|
||||
{
|
||||
_volumes = volumes;
|
||||
_indexing = indexing;
|
||||
_store = store;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public async Task RefreshAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
var sources = await _store.Sources.GetAllAsync(cancellationToken).ConfigureAwait(false);
|
||||
lock (_gate)
|
||||
{
|
||||
foreach (var source in sources)
|
||||
{
|
||||
var online = source.LastRootPath is not null && _volumes.IsPathReachable(source.LastRootPath);
|
||||
if (online && source.IsIndexed)
|
||||
{
|
||||
EnsureWatcher(source);
|
||||
}
|
||||
else
|
||||
{
|
||||
RemoveWatcher(source.Id);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void EnsureWatcher(Source source)
|
||||
{
|
||||
if (_watchers.ContainsKey(source.Id) || source.LastRootPath is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
var watcher = new FileSystemWatcher(source.LastRootPath)
|
||||
{
|
||||
IncludeSubdirectories = true,
|
||||
NotifyFilter = NotifyFilters.FileName | NotifyFilters.DirectoryName | NotifyFilters.Size | NotifyFilters.LastWrite | NotifyFilters.Attributes,
|
||||
InternalBufferSize = 64 * 1024
|
||||
};
|
||||
var sourceId = source.Id;
|
||||
void OnChange(object s, FileSystemEventArgs e)
|
||||
{
|
||||
var rel = PathRules.MakeRelative(source.LastRootPath!, PathRules.Parent(e.FullPath));
|
||||
_indexing.EnqueueReconcile(sourceId, rel);
|
||||
}
|
||||
|
||||
watcher.Created += OnChange;
|
||||
watcher.Changed += OnChange;
|
||||
watcher.Deleted += OnChange;
|
||||
watcher.Renamed += OnChange;
|
||||
watcher.Error += (_, args) =>
|
||||
{
|
||||
_logger.LogWarning(args.GetException(), "Watcher overflow for source {Id}", sourceId);
|
||||
_ = _store.Sources.UpdateStatusAsync(sourceId, SourceStatus.Stale, "Filesystem watcher overflow");
|
||||
};
|
||||
watcher.EnableRaisingEvents = true;
|
||||
_watchers[source.Id] = watcher;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogDebug(ex, "Watcher not available for {Path}", source.LastRootPath);
|
||||
}
|
||||
}
|
||||
|
||||
private void RemoveWatcher(long sourceId)
|
||||
{
|
||||
if (_watchers.Remove(sourceId, out var watcher))
|
||||
{
|
||||
watcher.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
foreach (var w in _watchers.Values)
|
||||
{
|
||||
w.Dispose();
|
||||
}
|
||||
|
||||
_watchers.Clear();
|
||||
}
|
||||
}
|
||||
}
|
||||
13
src/Explorer.Indexing/Explorer.Indexing.csproj
Normal file
13
src/Explorer.Indexing/Explorer.Indexing.csproj
Normal file
@@ -0,0 +1,13 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
<PropertyGroup>
|
||||
<RootNamespace>Explorer.Indexing</RootNamespace>
|
||||
</PropertyGroup>
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Microsoft.Extensions.Hosting.Abstractions" Version="10.0.0" />
|
||||
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.0" />
|
||||
</ItemGroup>
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\Explorer.Application\Explorer.Application.csproj" />
|
||||
<ProjectReference Include="..\Explorer.Domain\Explorer.Domain.csproj" />
|
||||
</ItemGroup>
|
||||
</Project>
|
||||
309
src/Explorer.Indexing/FilesystemScanner.cs
Normal file
309
src/Explorer.Indexing/FilesystemScanner.cs
Normal file
@@ -0,0 +1,309 @@
|
||||
using Explorer.Application;
|
||||
using Explorer.Domain;
|
||||
using Explorer.Domain.Abstractions;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Explorer.Indexing;
|
||||
|
||||
public sealed class FilesystemScanner
|
||||
{
|
||||
private readonly IIndexStore _store;
|
||||
private readonly IFileSystemEnumerator _enumerator;
|
||||
private readonly StorageProviderRegistry _providers;
|
||||
private readonly ILogger<FilesystemScanner> _logger;
|
||||
|
||||
public FilesystemScanner(
|
||||
IIndexStore store,
|
||||
IFileSystemEnumerator enumerator,
|
||||
StorageProviderRegistry providers,
|
||||
ILogger<FilesystemScanner> logger)
|
||||
{
|
||||
_store = store;
|
||||
_enumerator = enumerator;
|
||||
_providers = providers;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public async Task<ScanJob> ScanAsync(
|
||||
Source source,
|
||||
ScanKind kind,
|
||||
string? folderPathRel,
|
||||
IProgress<ScanProgress>? progress,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
var generation = source.ScanGeneration + 1;
|
||||
source.ScanGeneration = generation;
|
||||
var job = new ScanJob
|
||||
{
|
||||
SourceId = source.Id,
|
||||
Kind = kind,
|
||||
Status = ScanJobStatus.Running,
|
||||
StartedUtc = now,
|
||||
FolderPathRel = folderPathRel
|
||||
};
|
||||
var pending = new List<IndexEntry>(AppConstants.ScanBatchSize);
|
||||
|
||||
try
|
||||
{
|
||||
await _store.Sources.UpsertAsync(source, CancellationToken.None).ConfigureAwait(false);
|
||||
await _store.Sources.UpdateStatusAsync(source.Id, SourceStatus.Scanning, null, CancellationToken.None)
|
||||
.ConfigureAwait(false);
|
||||
await _store.ScanJobs.InsertAsync(job, CancellationToken.None).ConfigureAwait(false);
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
|
||||
var excludes = await _store.Excludes.GetAllAsync(cancellationToken).ConfigureAwait(false);
|
||||
var evaluator = new ExcludeEvaluator(excludes);
|
||||
var root = source.LastRootPath ?? throw new InvalidOperationException("Source has no root path.");
|
||||
var startRel = folderPathRel ?? "";
|
||||
var startPath = PathRules.Combine(root, startRel);
|
||||
var lastProgress = DateTime.UtcNow;
|
||||
var visitedDirs = new HashSet<string>(StringComparer.OrdinalIgnoreCase);
|
||||
|
||||
var startItem = _enumerator.GetItem(startPath);
|
||||
long? startParentId = null;
|
||||
if (!string.IsNullOrEmpty(startRel))
|
||||
{
|
||||
var parentRel = PathRules.MakeRelative(root, PathRules.Parent(startPath));
|
||||
var parent = await _store.Entries.GetByPathAsync(source.Id, parentRel, cancellationToken).ConfigureAwait(false);
|
||||
startParentId = parent?.Id;
|
||||
}
|
||||
|
||||
var startEntry = CreateEntry(source, startParentId, startItem ?? DummyDir(startPath, startRel, root), startRel, now, generation);
|
||||
startEntry.IsDirectory = true;
|
||||
await _store.RunWriteAsync(async s =>
|
||||
{
|
||||
startEntry.Id = await s.Entries.UpsertAsync(startEntry, cancellationToken).ConfigureAwait(false);
|
||||
}, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
var stack = new Stack<Frame>();
|
||||
stack.Push(new Frame
|
||||
{
|
||||
Path = startPath,
|
||||
PathRel = startRel,
|
||||
Id = startEntry.Id,
|
||||
ParentId = startParentId
|
||||
});
|
||||
|
||||
while (stack.Count > 0)
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
var frame = stack.Peek();
|
||||
if (!frame.Expanded)
|
||||
{
|
||||
if (!visitedDirs.Add(frame.Path))
|
||||
{
|
||||
stack.Pop();
|
||||
continue;
|
||||
}
|
||||
|
||||
var children = await _providers.EnrichAsync(
|
||||
_enumerator.EnumerateChildrenSafe(frame.Path, out var error),
|
||||
cancellationToken).ConfigureAwait(false);
|
||||
if (error is not null)
|
||||
{
|
||||
job.ErrorCount++;
|
||||
job.LastError = error;
|
||||
await _store.ScanJobs.AddErrorAsync(new ScanError
|
||||
{
|
||||
JobId = job.Id,
|
||||
Path = frame.Path,
|
||||
Kind = error.Contains("denied", StringComparison.OrdinalIgnoreCase) ? "AccessDenied" : "Io",
|
||||
Message = error,
|
||||
Utc = DateTimeOffset.UtcNow
|
||||
}, cancellationToken).ConfigureAwait(false);
|
||||
frame.Expanded = true;
|
||||
continue;
|
||||
}
|
||||
|
||||
var childDirs = new List<Frame>();
|
||||
foreach (var child in children)
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
if (evaluator.ShouldExclude(child.FullPath, child.Name, child.IsDirectory, child.Attributes, source.Id))
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
var rel = PathRules.MakeRelative(root, child.FullPath);
|
||||
var entry = CreateEntry(source, frame.Id, child, rel, DateTimeOffset.UtcNow, generation);
|
||||
pending.Add(entry);
|
||||
if (child.IsDirectory)
|
||||
{
|
||||
job.DirsSeen++;
|
||||
if (ReparsePolicy.ShouldRecurseIntoDirectory(child))
|
||||
{
|
||||
childDirs.Add(new Frame
|
||||
{
|
||||
Path = child.FullPath,
|
||||
PathRel = rel,
|
||||
Parent = frame,
|
||||
ParentId = frame.Id
|
||||
});
|
||||
}
|
||||
else
|
||||
{
|
||||
frame.Dirs++;
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
job.FilesSeen++;
|
||||
job.BytesSeen += child.SizeBytes;
|
||||
frame.Size += child.SizeBytes;
|
||||
frame.Files++;
|
||||
}
|
||||
}
|
||||
|
||||
await FlushAsync(pending, childDirs, cancellationToken).ConfigureAwait(false);
|
||||
frame.Expanded = true;
|
||||
for (var i = childDirs.Count - 1; i >= 0; i--)
|
||||
{
|
||||
stack.Push(childDirs[i]);
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
await _store.Entries.UpdateAggregatesAsync(frame.Id, frame.Size, frame.Files, frame.Dirs, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
if (frame.Parent is not null)
|
||||
{
|
||||
frame.Parent.Size += frame.Size;
|
||||
frame.Parent.Dirs++;
|
||||
}
|
||||
|
||||
stack.Pop();
|
||||
}
|
||||
|
||||
if ((DateTime.UtcNow - lastProgress).TotalMilliseconds >= AppConstants.ProgressHzMilliseconds)
|
||||
{
|
||||
lastProgress = DateTime.UtcNow;
|
||||
progress?.Report(ToProgress(job, frame.Path));
|
||||
job.ResumePath = frame.PathRel;
|
||||
await _store.ScanJobs.UpdateAsync(job, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
await FlushAsync(pending, [], cancellationToken).ConfigureAwait(false);
|
||||
await _store.Entries.MarkMissingAsDeletedAsync(source.Id, generation, DateTimeOffset.UtcNow, string.IsNullOrEmpty(startRel) ? null : startRel, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
await _store.Sources.UpdateIndexedAsync(source.Id, DateTimeOffset.UtcNow, generation, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
|
||||
job.Status = ScanJobStatus.Done;
|
||||
job.FinishedUtc = DateTimeOffset.UtcNow;
|
||||
await _store.ScanJobs.UpdateAsync(job, cancellationToken).ConfigureAwait(false);
|
||||
progress?.Report(ToProgress(job, startPath) with { Status = ScanJobStatus.Done });
|
||||
return job;
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
await FlushAsync(pending, [], CancellationToken.None).ConfigureAwait(false);
|
||||
job.Status = ScanJobStatus.Cancelled;
|
||||
job.FinishedUtc = DateTimeOffset.UtcNow;
|
||||
await _store.ScanJobs.UpdateAsync(job, CancellationToken.None).ConfigureAwait(false);
|
||||
await _store.Sources.UpdateStatusAsync(source.Id, SourceStatus.Stale, "Scan cancelled", CancellationToken.None)
|
||||
.ConfigureAwait(false);
|
||||
throw;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Scan failed for source {Id}", source.Id);
|
||||
job.Status = ScanJobStatus.Failed;
|
||||
job.LastError = ex.Message;
|
||||
job.FinishedUtc = DateTimeOffset.UtcNow;
|
||||
await _store.ScanJobs.UpdateAsync(job, CancellationToken.None).ConfigureAwait(false);
|
||||
await _store.Sources.UpdateStatusAsync(source.Id, SourceStatus.Error, ex.Message, CancellationToken.None)
|
||||
.ConfigureAwait(false);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
private async Task FlushAsync(List<IndexEntry> pending, List<Frame> dirs, CancellationToken cancellationToken)
|
||||
{
|
||||
if (pending.Count == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
await _store.RunWriteAsync(async s =>
|
||||
{
|
||||
foreach (var entry in pending)
|
||||
{
|
||||
entry.Id = await s.Entries.UpsertAsync(entry, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
foreach (var d in dirs)
|
||||
{
|
||||
var match = pending.Find(e =>
|
||||
e.IsDirectory && string.Equals(e.PathRel, d.PathRel, StringComparison.OrdinalIgnoreCase));
|
||||
if (match is not null)
|
||||
{
|
||||
d.Id = match.Id;
|
||||
}
|
||||
}
|
||||
|
||||
pending.Clear();
|
||||
}
|
||||
|
||||
private static ScanProgress ToProgress(ScanJob job, string path) => new()
|
||||
{
|
||||
JobId = job.Id,
|
||||
SourceId = job.SourceId,
|
||||
CurrentPath = path,
|
||||
FilesSeen = job.FilesSeen,
|
||||
DirsSeen = job.DirsSeen,
|
||||
BytesSeen = job.BytesSeen,
|
||||
ErrorCount = job.ErrorCount,
|
||||
Status = job.Status
|
||||
};
|
||||
|
||||
private static FileSystemItem DummyDir(string startPath, string startRel, string root) => new()
|
||||
{
|
||||
FullPath = startPath,
|
||||
Name = string.IsNullOrEmpty(startRel) ? root.TrimEnd('\\') : PathRules.GetFileName(startPath),
|
||||
IsDirectory = true,
|
||||
SizeBytes = 0,
|
||||
Attributes = AttributeFlags.Directory,
|
||||
ReparseTag = 0
|
||||
};
|
||||
|
||||
private static IndexEntry CreateEntry(Source source, long? parentId, FileSystemItem item, string pathRel, DateTimeOffset now, long generation)
|
||||
=> new()
|
||||
{
|
||||
SourceId = source.Id,
|
||||
ParentId = parentId,
|
||||
Name = item.Name,
|
||||
NameNorm = NameNormalizer.Normalize(item.Name),
|
||||
Extension = item.IsDirectory ? null : NameNormalizer.Extension(item.Name),
|
||||
IsDirectory = item.IsDirectory,
|
||||
SizeBytes = item.IsDirectory ? 0 : item.SizeBytes,
|
||||
AggregateSize = item.IsDirectory ? 0 : item.SizeBytes,
|
||||
CreatedUtc = item.CreatedUtc,
|
||||
ModifiedUtc = item.ModifiedUtc,
|
||||
LastSeenUtc = now,
|
||||
LastIndexedUtc = now,
|
||||
Attributes = item.Attributes,
|
||||
FileId = item.FileId,
|
||||
ReparseTag = item.ReparseTag,
|
||||
Status = EntryStatus.Present,
|
||||
PathRel = pathRel,
|
||||
ScanGeneration = generation,
|
||||
AllocatedSizeBytes = item.AllocatedSizeBytes ?? item.Cloud?.AllocatedSizeBytes,
|
||||
CloudAvailability = item.Cloud?.Availability
|
||||
};
|
||||
|
||||
private sealed class Frame
|
||||
{
|
||||
public required string Path { get; init; }
|
||||
public required string PathRel { get; init; }
|
||||
public long Id { get; set; }
|
||||
public long? ParentId { get; init; }
|
||||
public Frame? Parent { get; init; }
|
||||
public bool Expanded { get; set; }
|
||||
public long Size { get; set; }
|
||||
public int Files { get; set; }
|
||||
public int Dirs { get; set; }
|
||||
}
|
||||
}
|
||||
106
src/Explorer.Indexing/FolderReconciler.cs
Normal file
106
src/Explorer.Indexing/FolderReconciler.cs
Normal file
@@ -0,0 +1,106 @@
|
||||
using Explorer.Application;
|
||||
using Explorer.Domain;
|
||||
using Explorer.Domain.Abstractions;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Explorer.Indexing;
|
||||
|
||||
public sealed class FolderReconciler
|
||||
{
|
||||
private readonly IIndexStore _store;
|
||||
private readonly IFileSystemEnumerator _enumerator;
|
||||
private readonly StorageProviderRegistry _providers;
|
||||
|
||||
public FolderReconciler(IIndexStore store, IFileSystemEnumerator enumerator, StorageProviderRegistry providers)
|
||||
{
|
||||
_store = store;
|
||||
_enumerator = enumerator;
|
||||
_providers = providers;
|
||||
}
|
||||
|
||||
public async Task ReconcileAsync(Source source, string pathRel, CancellationToken cancellationToken)
|
||||
{
|
||||
if (string.IsNullOrEmpty(source.LastRootPath))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var full = PathRules.Combine(source.LastRootPath, pathRel);
|
||||
var parent = await _store.Entries.GetByPathAsync(source.Id, pathRel, cancellationToken).ConfigureAwait(false);
|
||||
if (parent is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var live = await _providers.EnrichAsync(_enumerator.EnumerateChildrenSafe(full, out _), cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
var indexed = await _store.Entries.GetChildrenAsync(source.Id, parent.Id, EntryStatus.Present, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
var liveNames = new HashSet<string>(live.Select(i => NameNormalizer.Normalize(i.Name)), StringComparer.Ordinal);
|
||||
|
||||
await _store.RunWriteAsync(async s =>
|
||||
{
|
||||
foreach (var item in live)
|
||||
{
|
||||
var rel = PathRules.MakeRelative(source.LastRootPath, item.FullPath);
|
||||
var entry = new IndexEntry
|
||||
{
|
||||
SourceId = source.Id,
|
||||
ParentId = parent.Id,
|
||||
Name = item.Name,
|
||||
NameNorm = NameNormalizer.Normalize(item.Name),
|
||||
Extension = item.IsDirectory ? null : NameNormalizer.Extension(item.Name),
|
||||
IsDirectory = item.IsDirectory,
|
||||
SizeBytes = item.IsDirectory ? 0 : item.SizeBytes,
|
||||
AggregateSize = item.IsDirectory ? 0 : item.SizeBytes,
|
||||
CreatedUtc = item.CreatedUtc,
|
||||
ModifiedUtc = item.ModifiedUtc,
|
||||
LastSeenUtc = now,
|
||||
LastIndexedUtc = now,
|
||||
Attributes = item.Attributes,
|
||||
FileId = item.FileId,
|
||||
ReparseTag = item.ReparseTag,
|
||||
Status = EntryStatus.Present,
|
||||
PathRel = rel,
|
||||
ScanGeneration = source.ScanGeneration,
|
||||
AllocatedSizeBytes = item.AllocatedSizeBytes ?? item.Cloud?.AllocatedSizeBytes,
|
||||
CloudAvailability = item.Cloud?.Availability
|
||||
};
|
||||
|
||||
var existing = indexed.FirstOrDefault(e => e.NameNorm == entry.NameNorm);
|
||||
var oldSize = existing is { IsDirectory: false } ? existing.SizeBytes : 0;
|
||||
var id = await s.Entries.UpsertAsync(entry, cancellationToken).ConfigureAwait(false);
|
||||
if (!item.IsDirectory)
|
||||
{
|
||||
var delta = item.SizeBytes - oldSize;
|
||||
if (existing is null)
|
||||
{
|
||||
await s.Entries.ApplySizeDeltaToAncestorsAsync(parent.Id, item.SizeBytes, 1, 0, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
else if (delta != 0)
|
||||
{
|
||||
await s.Entries.ApplySizeDeltaToAncestorsAsync(parent.Id, delta, 0, 0, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
else if (existing is null)
|
||||
{
|
||||
await s.Entries.ApplySizeDeltaToAncestorsAsync(parent.Id, 0, 0, 1, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
_ = id;
|
||||
}
|
||||
|
||||
foreach (var old in indexed)
|
||||
{
|
||||
if (!liveNames.Contains(old.NameNorm))
|
||||
{
|
||||
await s.Entries.TombstoneAsync(old.Id, now, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
}, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
155
src/Explorer.Indexing/IndexingCoordinator.cs
Normal file
155
src/Explorer.Indexing/IndexingCoordinator.cs
Normal file
@@ -0,0 +1,155 @@
|
||||
using System.Threading.Channels;
|
||||
using Explorer.Domain;
|
||||
using Explorer.Domain.Abstractions;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Explorer.Indexing;
|
||||
|
||||
public sealed class IndexingCoordinator : BackgroundService
|
||||
{
|
||||
private readonly IIndexStore _store;
|
||||
private readonly FilesystemScanner _scanner;
|
||||
private readonly FolderReconciler _reconciler;
|
||||
private readonly UsnChangeApplier _usn;
|
||||
private readonly IUsnJournal _journal;
|
||||
private readonly IVolumeService _volumes;
|
||||
private readonly ILogger<IndexingCoordinator> _logger;
|
||||
private readonly Channel<IndexWork> _work = Channel.CreateUnbounded<IndexWork>();
|
||||
private readonly Dictionary<long, CancellationTokenSource> _running = new();
|
||||
private readonly object _gate = new();
|
||||
|
||||
public event EventHandler<ScanProgress>? ProgressChanged;
|
||||
|
||||
public IndexingCoordinator(
|
||||
IIndexStore store,
|
||||
FilesystemScanner scanner,
|
||||
FolderReconciler reconciler,
|
||||
UsnChangeApplier usn,
|
||||
IUsnJournal journal,
|
||||
IVolumeService volumes,
|
||||
ILogger<IndexingCoordinator> logger)
|
||||
{
|
||||
_store = store;
|
||||
_scanner = scanner;
|
||||
_reconciler = reconciler;
|
||||
_usn = usn;
|
||||
_journal = journal;
|
||||
_volumes = volumes;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public void EnqueueFullScan(long sourceId)
|
||||
=> _work.Writer.TryWrite(new IndexWork(WorkKind.Full, sourceId, null));
|
||||
|
||||
public void EnqueueFolderScan(long sourceId, string pathRel)
|
||||
=> _work.Writer.TryWrite(new IndexWork(WorkKind.Folder, sourceId, pathRel));
|
||||
|
||||
public void EnqueueReconcile(long sourceId, string pathRel)
|
||||
=> _work.Writer.TryWrite(new IndexWork(WorkKind.Reconcile, sourceId, pathRel));
|
||||
|
||||
public void EnqueueUsn(long sourceId)
|
||||
=> _work.Writer.TryWrite(new IndexWork(WorkKind.Usn, sourceId, null));
|
||||
|
||||
public void Cancel(long sourceId)
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
if (_running.TryGetValue(sourceId, out var cts))
|
||||
{
|
||||
cts.Cancel();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
await _store.ScanJobs.InterruptRunningAsync(stoppingToken).ConfigureAwait(false);
|
||||
_ = Task.Run(() => PeriodicAsync(stoppingToken), stoppingToken);
|
||||
|
||||
await foreach (var item in _work.Reader.ReadAllAsync(stoppingToken).ConfigureAwait(false))
|
||||
{
|
||||
try
|
||||
{
|
||||
await RunAsync(item, stoppingToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException) when (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
_logger.LogInformation("Indexing work cancelled for source {Id}", item.SourceId);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Indexing work failed for source {Id}", item.SourceId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async Task RunAsync(IndexWork item, CancellationToken stoppingToken)
|
||||
{
|
||||
var source = await _store.Sources.GetAsync(item.SourceId, stoppingToken).ConfigureAwait(false);
|
||||
if (source is null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
using var linked = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
|
||||
lock (_gate)
|
||||
{
|
||||
_running[item.SourceId] = linked;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
switch (item.Kind)
|
||||
{
|
||||
case WorkKind.Full:
|
||||
case WorkKind.Folder:
|
||||
await _scanner.ScanAsync(
|
||||
source,
|
||||
item.Kind == WorkKind.Full ? ScanKind.Full : ScanKind.Folder,
|
||||
item.PathRel,
|
||||
new Progress<ScanProgress>(p => ProgressChanged?.Invoke(this, p)),
|
||||
linked.Token).ConfigureAwait(false);
|
||||
EnqueueUsn(source.Id);
|
||||
break;
|
||||
case WorkKind.Reconcile:
|
||||
if (item.PathRel is not null)
|
||||
{
|
||||
await _reconciler.ReconcileAsync(source, item.PathRel, linked.Token).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
break;
|
||||
case WorkKind.Usn:
|
||||
await _usn.ApplyAsync(source, _journal, linked.Token).ConfigureAwait(false);
|
||||
break;
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
_running.Remove(item.SourceId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async Task PeriodicAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
using var timer = new PeriodicTimer(TimeSpan.FromSeconds(30));
|
||||
while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false))
|
||||
{
|
||||
var sources = await _store.Sources.GetAllAsync(stoppingToken).ConfigureAwait(false);
|
||||
foreach (var source in sources.Where(s => s.IsIndexed && s.Status is SourceStatus.Online or SourceStatus.Stale))
|
||||
{
|
||||
if (source.LastRootPath is not null && _volumes.IsPathReachable(source.LastRootPath))
|
||||
{
|
||||
EnqueueUsn(source.Id);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private enum WorkKind { Full, Folder, Reconcile, Usn }
|
||||
|
||||
private readonly record struct IndexWork(WorkKind Kind, long SourceId, string? PathRel);
|
||||
}
|
||||
180
src/Explorer.Indexing/UsnChangeApplier.cs
Normal file
180
src/Explorer.Indexing/UsnChangeApplier.cs
Normal file
@@ -0,0 +1,180 @@
|
||||
using Explorer.Domain;
|
||||
using Explorer.Domain.Abstractions;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace Explorer.Indexing;
|
||||
|
||||
public sealed class UsnChangeApplier
|
||||
{
|
||||
private readonly IIndexStore _store;
|
||||
private readonly IFileSystemEnumerator _enumerator;
|
||||
private readonly ILogger<UsnChangeApplier> _logger;
|
||||
|
||||
public UsnChangeApplier(IIndexStore store, IFileSystemEnumerator enumerator, ILogger<UsnChangeApplier> logger)
|
||||
{
|
||||
_store = store;
|
||||
_enumerator = enumerator;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
public async Task<UsnReadStatus> ApplyAsync(Source source, IUsnJournal journal, CancellationToken cancellationToken)
|
||||
{
|
||||
if (source.LastRootPath is null || source.Kind is not (SourceKind.NtfsLocal or SourceKind.Removable))
|
||||
{
|
||||
return UsnReadStatus.Unavailable;
|
||||
}
|
||||
|
||||
if (!journal.TryQuery(source.LastRootPath, out var current, out _))
|
||||
{
|
||||
return UsnReadStatus.Unavailable;
|
||||
}
|
||||
|
||||
var from = new UsnJournalState
|
||||
{
|
||||
JournalId = source.UsnJournalId ?? 0,
|
||||
NextUsn = source.UsnNext ?? 0
|
||||
};
|
||||
|
||||
if (from.JournalId != 0 && from.JournalId != current.JournalId)
|
||||
{
|
||||
await _store.Sources.UpdateStatusAsync(source.Id, SourceStatus.Stale, "Change journal reset", cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
return UsnReadStatus.JournalReset;
|
||||
}
|
||||
|
||||
if (from.JournalId == 0)
|
||||
{
|
||||
await _store.Sources.UpdateUsnAsync(source.Id, current.JournalId, current.NextUsn, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
return UsnReadStatus.Ok;
|
||||
}
|
||||
|
||||
var records = journal.Read(source.LastRootPath, from, 4000, out var next, out var status);
|
||||
if (status is UsnReadStatus.JournalReset)
|
||||
{
|
||||
await _store.Sources.UpdateStatusAsync(source.Id, SourceStatus.Stale, "Change journal reset", cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
return status;
|
||||
}
|
||||
|
||||
if (status != UsnReadStatus.Ok)
|
||||
{
|
||||
return status;
|
||||
}
|
||||
|
||||
var coalesced = Coalesce(records);
|
||||
var now = DateTimeOffset.UtcNow;
|
||||
await _store.RunWriteAsync(async s =>
|
||||
{
|
||||
foreach (var rec in coalesced)
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
try
|
||||
{
|
||||
await ApplyRecord(s, source, rec, now, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogDebug(ex, "USN apply failed for {Name}", rec.FileName);
|
||||
}
|
||||
}
|
||||
|
||||
await s.Sources.UpdateUsnAsync(source.Id, next.JournalId, next.NextUsn, cancellationToken).ConfigureAwait(false);
|
||||
}, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
return UsnReadStatus.Ok;
|
||||
}
|
||||
|
||||
private async Task ApplyRecord(IIndexStore store, Source source, UsnRecord rec, DateTimeOffset now, CancellationToken cancellationToken)
|
||||
{
|
||||
var existing = rec.FileReferenceNumber != 0
|
||||
? await store.Entries.GetByFileIdAsync(source.Id, rec.FileReferenceNumber, cancellationToken).ConfigureAwait(false)
|
||||
: null;
|
||||
|
||||
if (rec.IsDelete)
|
||||
{
|
||||
if (existing is not null)
|
||||
{
|
||||
await store.Entries.TombstoneAsync(existing.Id, now, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
var parent = rec.ParentFileReferenceNumber != 0
|
||||
? await store.Entries.GetByFileIdAsync(source.Id, rec.ParentFileReferenceNumber, cancellationToken).ConfigureAwait(false)
|
||||
: await store.Entries.GetRootAsync(source.Id, cancellationToken).ConfigureAwait(false);
|
||||
|
||||
var parentRel = parent?.PathRel ?? "";
|
||||
var rel = string.IsNullOrEmpty(parentRel) ? rec.FileName : parentRel + "\\" + rec.FileName;
|
||||
var full = PathRules.Combine(source.LastRootPath!, rel);
|
||||
var live = _enumerator.GetItem(full);
|
||||
|
||||
if (live is null)
|
||||
{
|
||||
if (existing is not null)
|
||||
{
|
||||
await store.Entries.TombstoneAsync(existing.Id, now, cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
var oldSize = existing is { IsDirectory: false } ? existing.SizeBytes : 0;
|
||||
var entry = new IndexEntry
|
||||
{
|
||||
SourceId = source.Id,
|
||||
ParentId = parent?.Id,
|
||||
Name = live.Name,
|
||||
NameNorm = NameNormalizer.Normalize(live.Name),
|
||||
Extension = live.IsDirectory ? null : NameNormalizer.Extension(live.Name),
|
||||
IsDirectory = live.IsDirectory,
|
||||
SizeBytes = live.IsDirectory ? 0 : live.SizeBytes,
|
||||
AggregateSize = existing?.AggregateSize ?? (live.IsDirectory ? 0 : live.SizeBytes),
|
||||
CreatedUtc = live.CreatedUtc,
|
||||
ModifiedUtc = live.ModifiedUtc,
|
||||
LastSeenUtc = now,
|
||||
LastIndexedUtc = now,
|
||||
Attributes = live.Attributes,
|
||||
FileId = rec.FileReferenceNumber,
|
||||
ParentFileId = rec.ParentFileReferenceNumber,
|
||||
ReparseTag = live.ReparseTag,
|
||||
Status = EntryStatus.Present,
|
||||
PathRel = rel,
|
||||
ScanGeneration = source.ScanGeneration
|
||||
};
|
||||
|
||||
if (existing is not null && !string.Equals(existing.PathRel, rel, StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
await store.Entries.RenameSubtreePathAsync(source.Id, existing.PathRel, rel, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
|
||||
await store.Entries.UpsertAsync(entry, cancellationToken).ConfigureAwait(false);
|
||||
if (!live.IsDirectory)
|
||||
{
|
||||
var delta = live.SizeBytes - oldSize;
|
||||
if (existing is null)
|
||||
{
|
||||
await store.Entries.ApplySizeDeltaToAncestorsAsync(parent?.Id, live.SizeBytes, 1, 0, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
else if (delta != 0)
|
||||
{
|
||||
await store.Entries.ApplySizeDeltaToAncestorsAsync(parent?.Id, delta, 0, 0, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static List<UsnRecord> Coalesce(IReadOnlyList<UsnRecord> records)
|
||||
{
|
||||
var map = new Dictionary<long, UsnRecord>();
|
||||
foreach (var rec in records)
|
||||
{
|
||||
map[rec.FileReferenceNumber] = rec;
|
||||
}
|
||||
|
||||
return map.Values.ToList();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user