From 812e04cbe3433afc272cda773ba4f2f4cf026579 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 27 Jul 2026 21:22:58 +0000 Subject: [PATCH 1/2] Document core synchronization flow --- src/WingSync.Core/Planning/WritePlanner.cs | 13 ++++++ src/WingSync.Core/Services/SyncCoordinator.cs | 42 +++++++++++++++++++ 2 files changed, 55 insertions(+) diff --git a/src/WingSync.Core/Planning/WritePlanner.cs b/src/WingSync.Core/Planning/WritePlanner.cs index 61c5a3a..1b65cda 100644 --- a/src/WingSync.Core/Planning/WritePlanner.cs +++ b/src/WingSync.Core/Planning/WritePlanner.cs @@ -305,6 +305,8 @@ private static bool FloatsEquivalent(float expected, float actual, float toleran return false; } + // WAPI may quantize the same control differently at small and large + // magnitudes, so accept either an absolute or a scale-relative match. var difference = MathF.Abs(expected - actual); var relativeScale = MathF.Max(MathF.Abs(expected), MathF.Abs(actual)); return difference <= tolerance @@ -327,6 +329,8 @@ public static WriteRisk Classify(SyncScope scope, TokenPath path) if (path.Segments.Count > 0 && path.Segments[0].Equals("io", StringComparison.OrdinalIgnoreCase)) { + // Physical I/O is always critical, even if a future catalog entry + // accidentally associates it with a less restrictive scope. return WriteRisk.Critical; } @@ -399,6 +403,8 @@ public WritePlan Plan( var inputOrder = 0; var superseded = 0; + // First turn each source observation into at most one policy-approved + // target candidate. Rejected observations remain visible as plan issues. foreach (var change in changes) { ArgumentNullException.ThrowIfNull(change); @@ -453,12 +459,17 @@ public WritePlan Plan( inputOrder++; } + // An unresolved sidechain reference makes the whole processor group + // unsafe: applying only its remaining leaves could create a mixed model. var candidates = candidatesByTarget.Values .Where(candidate => GetDynamicSourceGroup(candidate.SourcePath) is not { } sourceGroup || !blockedDynamicSourceGroups.Contains(sourceGroup)) .ToList(); AddModelSafetyGuards(candidates); + + // Preserve the first-seen order between independent processor groups, + // while enforcing guard -> model -> settings -> restore within a group. var groupOrder = candidates .GroupBy(static candidate => candidate.GroupKey, StringComparer.Ordinal) .ToDictionary( @@ -549,6 +560,8 @@ private static void AddModelSafetyGuards(ICollection candidates) continue; } + // Reuse the group's final enable value to synthesize a temporary + // off/bypass write; execution restores the original write last. foreach (var enablingWrite in group.Where(static candidate => candidate.IsEnablingWrite).ToArray()) { candidates.Add(enablingWrite.CreateGuard(nextSyntheticOrder++)); diff --git a/src/WingSync.Core/Services/SyncCoordinator.cs b/src/WingSync.Core/Services/SyncCoordinator.cs index 62fd2d1..1d120d8 100644 --- a/src/WingSync.Core/Services/SyncCoordinator.cs +++ b/src/WingSync.Core/Services/SyncCoordinator.cs @@ -102,8 +102,14 @@ public sealed class SyncCoordinator : IAsyncDisposable private readonly ISyncObserver observer; private readonly IClock clock; private readonly WritePlanner planner = new(); + + // Lifecycle operations serialize separately from reconciliation: Start/Stop + // own lifecycleGate, while every snapshot or write transaction owns reconcileGate. private readonly SemaphoreSlim lifecycleGate = new(1, 1); private readonly SemaphoreSlim reconcileGate = new(1, 1); + + // These mirrors are observational only. desiredTargets is the authoritative + // target intent used to detect drift after a plan has been accepted. private readonly ConcurrentDictionary sourceState = new(StringComparer.Ordinal); private readonly ConcurrentDictionary targetState = @@ -114,6 +120,9 @@ public sealed class SyncCoordinator : IAsyncDisposable new(StringComparer.Ordinal); private readonly ConcurrentDictionary echoFingerprints = new(StringComparer.Ordinal); + + // Event callbacks cannot await the reconciliation gate. They stage changes + // under this short lock until a coherent snapshot-to-stream hand-off occurs. private readonly object eventIntakeSync = new(); private readonly Dictionary deferredSourceEvents = new(StringComparer.Ordinal); @@ -129,6 +138,8 @@ public sealed class SyncCoordinator : IAsyncDisposable private readonly object latencySync = new(); private readonly Queue recentLatencies = new(); + // A session cancellation source owns one complete connection epoch and all + // worker tasks started for it. Reconnect replaces the transports, not the workers. private CancellationTokenSource? sessionCancellation; private Channel? workChannel; private Task? workerTask; @@ -244,6 +255,9 @@ public async Task StartAsync( liveWritesArmed = !configuration.Safety.DryRun && confirmLiveInitialSync; safetyPaused = false; sessionCancellation = new CancellationTokenSource(); + + // Multiple session callbacks produce work, but one consumer performs + // planning and writes so console mutations remain strictly ordered. workChannel = Channel.CreateBounded( new BoundedChannelOptions(QueueCapacity) { @@ -263,6 +277,8 @@ public async Task StartAsync( { ChangeStatus(SyncCoordinatorState.Connecting, "Identifying and connecting consoles."); + // Verify discovery identity before opening transports, then verify + // again through each connected console before any state is trusted. await VerifyIdentitiesAsync(cancellationToken).ConfigureAwait(false); await ConnectBothAsync(cancellationToken).ConfigureAwait(false); await VerifyConnectedIdentitiesAsync(cancellationToken).ConfigureAwait(false); @@ -283,6 +299,8 @@ public async Task StartAsync( workerTask = ProcessWorkAsync(sessionCancellation.Token); healthTask = HealthLoopAsync(sessionCancellation.Token); + // A live start without explicit approval stops at the exact fresh + // diff; event intake stays suspended so that preview cannot drift. if (!configuration.Safety.DryRun && !liveWritesArmed) { SuspendEventIntake(); @@ -395,6 +413,9 @@ pendingInitialPlan is null || ChangeStatus( SyncCoordinatorState.Snapshotting, "After confirmation, reread both consoles."); + + // Confirmation approves a diff, not merely an action. A changed diff + // must be shown again instead of inheriting the earlier approval. var freshSnapshot = await BuildFreshPlanAsync(cancellationToken).ConfigureAwait(false); var freshPlan = freshSnapshot.Plan; RecordPlanningFindings(freshPlan); @@ -783,6 +804,8 @@ private async Task BuildFreshPlanAsync(CancellationToken canc var wasInvalidated = false; for (var attempt = 1; attempt <= SnapshotStabilityAttempts; attempt++) { + // Sequence fences cover both console callbacks and transport changes. + // Only an attempt with identical fences can authorize live execution. var before = GetEventSequences(); var plan = await ReadFreshPlanAttemptAsync(cancellationToken).ConfigureAwait(false); var after = GetEventSequences(); @@ -821,6 +844,8 @@ private async Task ReadFreshPlanAttemptAsync(CancellationToken cancel readbackWaiters.Clear(); echoFingerprints.Clear(); + // Read both consoles concurrently to minimize the interval in which an + // operator action could make the two snapshots describe different moments. var sourceNodes = GetSnapshotNodes(currentConfiguration, sourceSide: true); var targetNodes = GetSnapshotNodes(currentConfiguration, sourceSide: false); var sourceSnapshotTask = ReadNodesAsync( @@ -906,6 +931,8 @@ private async Task> ReadNodesAsync( { for (var attempt = 1; attempt <= 3; attempt++) { + // Two scalar-certified passes must agree. Node snapshots alone are + // insufficient because some firmware exposes a stale local mirror. var first = await ReadScalarCertifiedNodePassAsync( session, nodes, @@ -944,6 +971,9 @@ private async Task CancellationToken cancellationToken) { var candidates = new Dictionary(StringComparer.Ordinal); + + // Node reads discover the available leaves efficiently; exact scalar reads + // below certify every value that can enter the plan. foreach (var node in nodes) { var snapshot = await session.SnapshotAsync(node.ToString(), cancellationToken) @@ -1047,6 +1077,8 @@ private WritePlan FilterUnchangedWrites(WritePlan completePlan) .Where(static group => group is not null) .ToHashSet(StringComparer.Ordinal); + // Model changes can reset sibling leaves. Keep the entire guarded group, + // including values already equal on the target, so it is rebuilt coherently. var retained = new List(); foreach (var write in completePlan.Writes) { @@ -2411,6 +2443,8 @@ private async Task ProcessWorkAsync(CancellationToken cancellationToken) continue; } + // The short window absorbs bursts from one physical control move. + // Last source value wins; source work outranks target-drift work. await clock.Delay(TimeSpan.FromMilliseconds(25), cancellationToken).ConfigureAwait(false); var coalesced = new Dictionary(StringComparer.Ordinal) { @@ -2455,6 +2489,8 @@ private async Task ProcessWorkAsync(CancellationToken cancellationToken) IReadOnlyList? expanded = null; for (var attempt = 1; attempt <= 3; attempt++) { + // A model event may imply changes to sibling settings. + // Expand and accept the group only while its event fence holds. var expansionFence = GetEventSequences(); var candidate = await ExpandModelChangesAsync(valid, cancellationToken) .ConfigureAwait(false); @@ -4112,6 +4148,9 @@ private async Task ReconnectLoopAsync(string initialReason, CancellationToken ca while (!cancellationToken.IsCancellationRequested && !stopping) { attempt++; + + // Cap exponential backoff so recovery remains responsive without + // creating a tight retry loop during an extended outage. var delay = TimeSpan.FromMilliseconds( Math.Min( MaximumReconnectDelay.TotalMilliseconds, @@ -4132,6 +4171,9 @@ private async Task ReconnectLoopAsync(string initialReason, CancellationToken ca await VerifyConnectedIdentitiesAsync(cancellationToken).ConfigureAwait(false); await BeginCacheEpochAsync(cancellationToken).ConfigureAwait(false); ChangeStatus(SyncCoordinatorState.Snapshotting, "After reconnect, reread both consoles."); + + // Reconnection never resumes from cached state. It establishes + // a new epoch and rebuilds target intent from fresh console reads. var freshSnapshot = await BuildFreshPlanAsync(cancellationToken).ConfigureAwait(false); while (!IsSnapshotCurrent(freshSnapshot)) { From d175ee4dc0ec690a600f7c7dd960dab4c8d935ca Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 27 Jul 2026 21:24:07 +0000 Subject: [PATCH 2/2] Document infrastructure and UI flows --- native/WapiHost/main.cpp | 9 +++++++++ src/WingSync.App/ViewModels/MainViewModel.cs | 14 ++++++++++++++ src/WingSync.Core/Planning/ScopeCatalog.cs | 6 ++++++ src/WingSync.Core/Planning/TokenPath.cs | 6 ++++++ .../Discovery/WingAnnouncementParser.cs | 4 ++++ .../Discovery/WingDiscoveryService.cs | 10 ++++++++++ .../Persistence/AtomicJsonFile.cs | 7 +++++++ .../Persistence/ConfigStore.cs | 4 ++++ .../Persistence/WingStateCache.cs | 13 +++++++++++++ .../Wapi/WapiProcessSession.cs | 12 ++++++++++++ 10 files changed, 85 insertions(+) diff --git a/native/WapiHost/main.cpp b/native/WapiHost/main.cpp index 81fecae..355e504 100644 --- a/native/WapiHost/main.cpp +++ b/native/WapiHost/main.cpp @@ -600,6 +600,8 @@ bool parseSetBatch( std::vector seenTokens; seenTokens.reserve(itemCount); + // Validate the complete batch before constructing any WAPI updates. A single + // malformed or duplicate token rejects the command without a partial write. for (std::size_t itemIndex = 0; itemIndex < itemCount; ++itemIndex) { const std::size_t fieldIndex = 2U + itemIndex * 3U; @@ -820,6 +822,8 @@ class WapiHost return; } + // Keepalive and event draining share the host thread with commands. This + // preserves the vendor library's single-threaded call boundary. const int keepAliveResult = wKeepAlive(); if (keepAliveResult != WSUCCESS && keepAliveResult != WZERO) { @@ -1440,6 +1444,9 @@ int runHost() CommandQueue queue; std::atomic_bool stopReader = false; + + // A dedicated reader prevents blocking stdin from starving WAPI keepalive and + // event polling. It never calls the vendor library; it only fills this queue. std::thread reader([&queue, &stopReader] { std::string line; @@ -1488,6 +1495,8 @@ int runHost() inputEnded = queue.inputEnded && queue.lines.empty(); } + // Limit command work per turn so a busy managed client cannot starve + // console events indefinitely. for (std::string& line : pending) { if (!host.handle(std::move(line))) diff --git a/src/WingSync.App/ViewModels/MainViewModel.cs b/src/WingSync.App/ViewModels/MainViewModel.cs index 0c73d97..fe68376 100644 --- a/src/WingSync.App/ViewModels/MainViewModel.cs +++ b/src/WingSync.App/ViewModels/MainViewModel.cs @@ -1005,6 +1005,8 @@ private async Task InitializeAsync() } else { + // First run and failed recovery both start from a deliberately small, + // dry-run-safe topology rather than inferring a live configuration. ChannelMappings.Add(new ChannelMappingViewModel(1, 1, "Channel 1")); AttachMappingHandlers(); ApplySafeDefaultScopes(); @@ -1058,6 +1060,9 @@ private async Task DiscoverAsync(bool silent) var previousStageSerial = SelectedStageWing?.Wing.SerialNumber ?? activeConfiguration?.Monitor.ExpectedSerial; + + // Rebuild the collection but restore assignments by pinned identity; + // list position and IP address may change between discovery passes. DiscoveredWings.Clear(); foreach (var wing in result.Wings) { @@ -1314,6 +1319,8 @@ private async Task StartAsync() try { + // Persist a safe projection before starting. Session-only live and + // high-risk approvals must never survive an application restart. await configStore.SaveAsync(CreatePersistedSafeConfiguration(configuration)); activeConfiguration = configuration; ClearProblem(); @@ -1347,6 +1354,8 @@ private async Task ReviewAndConfirmLiveInitialSyncAsync(AppConfiguration configu const int MaximumPreviewChanges = 3; for (var attempt = 1; attempt <= MaximumPreviewChanges; attempt++) { + // Each loop reviews one immutable snapshot. If either console changes, + // the coordinator replaces it and this method asks the operator again. var preview = coordinator.PendingInitialPreview ?? throw new InvalidOperationException( "Live synchronization was waiting without a valid fresh diff preview."); @@ -1710,6 +1719,8 @@ private void NotifyTopologyChanged() { if (!suppressLiveSafetyReset) { + // Any topology edit invalidates evidence gathered for the previous + // setup and revokes volatile live/high-risk consent. connectionTestSucceeded = false; hasCompletedDryRun = false; var resetToSafeMode = false; @@ -1987,6 +1998,9 @@ diagnosticEvent.Properties is null || diagnosticEvent.Code == "PARAMETER_BLOCKED" ? "Blocked" : "Executable")); + + // Diagnostics can arrive off the UI thread, so retain preview details in + // a bounded concurrent queue and materialize them only when opening a dialog. var count = Interlocked.Increment(ref previewTokenAuditCount); while (count > MaximumPreviewAuditEntries && previewTokenAudit.TryDequeue(out _)) diff --git a/src/WingSync.Core/Planning/ScopeCatalog.cs b/src/WingSync.Core/Planning/ScopeCatalog.cs index 182c84d..3adf4f1 100644 --- a/src/WingSync.Core/Planning/ScopeCatalog.cs +++ b/src/WingSync.Core/Planning/ScopeCatalog.cs @@ -121,6 +121,8 @@ public static bool TryMatch(TokenPath path, out SyncScope scope) var segments = path.Segments; var section = segments[2]; + // Three-segment channel leaves are controls directly on the channel. + // Deeper paths are matched by their processing subtree below. if (segments.Count == 3) { if (CustomLeaves.Contains(section)) @@ -613,6 +615,8 @@ public static bool TryRewrite( if (source.Type == WingValueType.I) { + // Native integer sidechain values encode regular input channels only; + // values outside that range represent non-channel choices and pass through. var integerSourceChannel = source.AsInt32(); if (integerSourceChannel is < WingChannelLimits.FirstInput or > WingChannelLimits.LastInput) { @@ -637,6 +641,8 @@ public static bool TryRewrite( out var prefix, out var separator)) { + // External sources, self, off, and other named choices are semantic + // values rather than channel addresses and must remain unchanged. return true; } diff --git a/src/WingSync.Core/Planning/TokenPath.cs b/src/WingSync.Core/Planning/TokenPath.cs index ccf4c0f..db72138 100644 --- a/src/WingSync.Core/Planning/TokenPath.cs +++ b/src/WingSync.Core/Planning/TokenPath.cs @@ -57,6 +57,9 @@ public static bool TryParse(string? value, out TokenPath result) } var canonical = value.Trim(); + + // Require a single rooted representation. Rejecting alternate spellings + // keeps dictionary keys, fingerprints, and scope matching deterministic. if (canonical.Length < 2 || canonical[0] != '/' || canonical[^1] == '/' @@ -66,6 +69,9 @@ public static bool TryParse(string? value, out TokenPath result) } var parsedSegments = canonical[1..].Split('/'); + + // Dot segments and whitespace could change meaning across consumers, so + // they are rejected rather than normalized silently. if (parsedSegments.Length == 0 || parsedSegments.Any(static segment => segment.Length == 0 diff --git a/src/WingSync.Infrastructure/Discovery/WingAnnouncementParser.cs b/src/WingSync.Infrastructure/Discovery/WingAnnouncementParser.cs index b108169..93461e1 100644 --- a/src/WingSync.Infrastructure/Discovery/WingAnnouncementParser.cs +++ b/src/WingSync.Infrastructure/Discovery/WingAnnouncementParser.cs @@ -45,6 +45,8 @@ public static bool TryParse( var announcement = Encoding.ASCII.GetString(payload); var fields = announcement.Split(',', StringSplitOptions.None); + // Trust the datagram source, not merely advertised text: requiring the + // two addresses to match prevents stale or forwarded announcements. if (fields.Length != 6 || !string.Equals(fields[0], "WING", StringComparison.Ordinal) || !TryParseCanonicalIPv4(fields[1], out var advertisedAddress) @@ -92,6 +94,8 @@ private static bool TryParseCanonicalIPv4( || address.Equals(IPAddress.Broadcast) || IsIPv4Multicast(address)) { + // Canonical text rejects ambiguous shorthand such as leading-zero + // octets before the address is persisted as a console identity. address = null; return false; } diff --git a/src/WingSync.Infrastructure/Discovery/WingDiscoveryService.cs b/src/WingSync.Infrastructure/Discovery/WingDiscoveryService.cs index 70e9754..3ce70ce 100644 --- a/src/WingSync.Infrastructure/Discovery/WingDiscoveryService.cs +++ b/src/WingSync.Infrastructure/Discovery/WingDiscoveryService.cs @@ -58,6 +58,8 @@ public async Task DiscoverAsync( CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); timeoutSource.CancelAfter(options.ReplyTimeout); + // Probe adapters concurrently and keep listening for the complete reply + // window; consoles may answer at different speeds on different subnets. var adapterTasks = bindings.Select( binding => QueryBindingAsync( binding, @@ -102,6 +104,9 @@ private static List GetBindings( } var targets = new HashSet(); + + // Directed broadcast reaches the adapter subnet reliably; + // the global address is an optional fallback and is deduplicated. if (options.IncludeDirectedBroadcast && TryCalculateDirectedBroadcast( unicast.Address, @@ -167,6 +172,8 @@ private async Task QueryBindingAsync( issues, cancellationToken); + // Start receiving before sending so a fast local reply cannot arrive + // between probe transmission and listener setup. foreach (var target in binding.Targets) { try @@ -229,6 +236,9 @@ private async Task ReceiveRepliesAsync( wing.SerialNumber, "\n", wing.IpAddress); + + // A console may answer both directed and global broadcasts. Serial + // plus address identifies one announcement without hiding duplicate IPs. discovered.TryAdd(key, wing); } } diff --git a/src/WingSync.Infrastructure/Persistence/AtomicJsonFile.cs b/src/WingSync.Infrastructure/Persistence/AtomicJsonFile.cs index 84334f5..b6cdc0c 100644 --- a/src/WingSync.Infrastructure/Persistence/AtomicJsonFile.cs +++ b/src/WingSync.Infrastructure/Persistence/AtomicJsonFile.cs @@ -20,6 +20,8 @@ public static async Task WriteAsync( try { + // Fully persist a same-directory temporary file before making it + // visible. Keeping it on the same volume makes replacement atomic. await using (var stream = new FileStream( temporaryPath, FileMode.CreateNew, @@ -36,6 +38,8 @@ public static async Task WriteAsync( cancellationToken.ThrowIfCancellationRequested(); if (File.Exists(targetPath)) { + // File.Replace atomically promotes the new bytes and rotates the + // previous primary into the caller-provided recovery path. File.Replace( temporaryPath, targetPath, @@ -75,6 +79,9 @@ public static async Task ReadBoundedAsync( using var content = new MemoryStream( capacity: (int)Math.Min(stream.Length, maximumBytes)); var buffer = new byte[16_384]; + + // Recheck the bound while streaming because another process may append + // after the initial length check. while (true) { var read = await stream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false); diff --git a/src/WingSync.Infrastructure/Persistence/ConfigStore.cs b/src/WingSync.Infrastructure/Persistence/ConfigStore.cs index 1a0bfc3..f92fb6e 100644 --- a/src/WingSync.Infrastructure/Persistence/ConfigStore.cs +++ b/src/WingSync.Infrastructure/Persistence/ConfigStore.cs @@ -95,6 +95,8 @@ public async Task LoadAsync( string? quarantinedPath = null; if (primary.Status == FileLoadStatus.Corrupt) { + // Preserve malformed content for support inspection, but move it + // out of the active path before attempting backup recovery. if (!TryQuarantine( _filePath, "corrupt", @@ -112,6 +114,8 @@ public async Task LoadAsync( .ConfigureAwait(false); if (backup.Status == FileLoadStatus.Loaded) { + // Recovery is reported to the caller rather than silently + // rewriting the primary; the next explicit save performs rotation. return new ConfigLoadResult( ConfigLoadStatus.RecoveredFromBackup, backup.Configuration, diff --git a/src/WingSync.Infrastructure/Persistence/WingStateCache.cs b/src/WingSync.Infrastructure/Persistence/WingStateCache.cs index 1bac7d4..b32c309 100644 --- a/src/WingSync.Infrastructure/Persistence/WingStateCache.cs +++ b/src/WingSync.Infrastructure/Persistence/WingStateCache.cs @@ -79,6 +79,8 @@ public void QueueWrite(WingStateSnapshot snapshot) ObjectDisposedException.ThrowIf(_disposed, this); if (_pending.Remove(key, out var previous)) { + // One partition keeps only its newest observation. Cancelling + // the old delay prevents a stale snapshot from winning the race. previous.DelayCancellation.Cancel(); } @@ -155,6 +157,8 @@ public async Task LoadAsync( var mismatch = primary.Status == CacheFileStatus.IdentityMismatch; if (primary.Status is CacheFileStatus.Corrupt or CacheFileStatus.IdentityMismatch) { + // Remove an untrustworthy primary before considering its backup; + // this prevents the same bad file from being retried at next start. if (!TryQuarantine( primaryPath, mismatch ? "identity-mismatch" : "corrupt", @@ -325,6 +329,8 @@ public async Task ResetAsync( await _resetGate.WaitAsync(cancellationToken).ConfigureAwait(false); try { + // Reset excludes QueueWrite for the entire cancel/drain/move sequence, + // so no debounce can recreate a file in the directory being emptied. PendingWrite[] pendingWrites; lock (_pendingLock) { @@ -489,6 +495,9 @@ JsonException or var primaryPath = Path.GetFullPath(GetPrimaryPath(envelope.Identity)); var candidatePath = Path.GetFullPath(path); + + // Accept only files whose content-derived partition maps back to + // their actual filename; copied or renamed cache files are ignored. if (!candidatePath.Equals(primaryPath, StringComparison.OrdinalIgnoreCase) && !candidatePath.Equals( string.Concat(primaryPath, ".bak"), @@ -578,6 +587,8 @@ await PersistSnapshotAsync(pending.Snapshot, cancellationToken) private async Task FlushCoreAsync(CancellationToken cancellationToken) { + // QueueWrite can add work while an earlier batch is being persisted. + // Repeat until a lock-protected observation finds no pending generations. while (true) { PendingWrite[] writes; @@ -630,6 +641,8 @@ private async Task PersistSnapshotAsync( var primaryPath = GetPrimaryPath(snapshot.Identity); var backupPath = string.Concat(primaryPath, ".bak"); + // Serialize all disk access so primary/backup rotation cannot interleave + // with a load, reset, or another partition write. await _ioGate.WaitAsync(cancellationToken).ConfigureAwait(false); try { diff --git a/src/WingSync.Infrastructure/Wapi/WapiProcessSession.cs b/src/WingSync.Infrastructure/Wapi/WapiProcessSession.cs index abcd536..c098b5f 100644 --- a/src/WingSync.Infrastructure/Wapi/WapiProcessSession.cs +++ b/src/WingSync.Infrastructure/Wapi/WapiProcessSession.cs @@ -21,8 +21,14 @@ public sealed class WapiProcessSession : IWingSession private readonly ISyncObserver observer; private readonly IClock clock; private readonly TimeSpan commandTimeout; + + // Process lifecycle and wire-command ownership are distinct. A lifecycle + // change may stop a helper only after the single FIFO command lease is resolved. private readonly SemaphoreSlim lifecycleGate = new(1, 1); private readonly SemaphoreSlim commandGate = new(1, 1); + + // Request IDs correlate asynchronous stdout frames with the command that + // owns them; the process generation rejects late frames from a replaced helper. private readonly ConcurrentDictionary pending = new(StringComparer.Ordinal); private readonly CancellationTokenSource lifetimeCancellation = new(); private Process? process; @@ -291,6 +297,8 @@ private async Task StartProcessAsync(CancellationToken cancellationToken) }; startInfo.Environment["WINGSYNC_ROLE"] = role; + // Publish the generation before starting readers so every callback can + // prove that it belongs to this exact helper process. var generation = Interlocked.Increment(ref processGeneration); Volatile.Write(ref activeProcessGeneration, generation); processCancellation = CancellationTokenSource.CreateLinkedTokenSource( @@ -506,6 +514,8 @@ private void HandleOutputLine( return; } + // Stdout is reserved for this framed protocol. Values and free-form + // details are Base64 encoded so the pipe separator remains unambiguous. var parts = line.Split('|'); if (parts.Length == 0) { @@ -617,6 +627,8 @@ private void CompleteSnapshot(string id, string countText, long generation) if (count != request.Items.Count) { + // END is the snapshot commit marker. A mismatched count means the + // stream was truncated and none of its partial values may be trusted. request.SnapshotCompletion.TrySetException( new InvalidDataException( $"Snapshot was incomplete: helper reported {count}, received {request.Items.Count}."));