From 00f14682ef03169f4888c32f7f38e241363d9f02 Mon Sep 17 00:00:00 2001 From: Kevin Hahn Date: Mon, 26 Oct 2026 14:53:32 +0700 Subject: [PATCH] Make JsonSyncable tolerate an interrupted write A client file can end in a partial line when a drive is pulled mid-sync. Reads now skip an unterminated last line that fails to parse, and appends first truncate that fragment (or terminate it, if it is a complete commit) so new commits land on their own line. Malformed lines elsewhere still throw. Co-Authored-By: Claude Opus 5.5 --- src/SIL.Harmony.Tests/JsonSyncableTests.cs | 196 +++++++++++++++++++++ src/SIL.Harmony/JsonSyncable.cs | 124 ++++++++++++- 2 files changed, 313 insertions(+), 7 deletions(-) diff --git a/src/SIL.Harmony.Tests/JsonSyncableTests.cs b/src/SIL.Harmony.Tests/JsonSyncableTests.cs index 26c3ef28..e63f981d 100644 --- a/src/SIL.Harmony.Tests/JsonSyncableTests.cs +++ b/src/SIL.Harmony.Tests/JsonSyncableTests.cs @@ -75,4 +75,200 @@ public async Task DoesNotDuplicateFileLines() dir.Delete(recursive: true); } } + + private static Commit MakeCommit(Guid clientId, int secondsOffset) => new() + { + ClientId = clientId, + HybridDateTime = new HybridDateTime(new DateTimeOffset(2024, 1, 1, 0, 0, 0, TimeSpan.Zero).AddSeconds(secondsOffset), 0), + ChangeEntities = + { + new ChangeEntity + { + Change = new SetWordTextChange(Guid.NewGuid(), "hello"), + Index = 0, + CommitId = Guid.Empty, + EntityId = Guid.Empty + } + } + }; + + private static string ClientFile(DirectoryInfo dir, Guid clientId) => + Path.Combine(dir.FullName, $"{JsonSyncable.FilenamePrefix}{clientId}{JsonSyncable.FilenameExtension}"); + + private static string Serialize(Commit commit) => JsonSerializer.Serialize(commit, SerializerOptions); + + //simulates a write interrupted partway through a line, e.g. a USB drive pulled mid-sync + private static string Truncated(Commit commit) + { + var json = Serialize(commit); + return json[..(json.Length / 2)]; + } + + private static SyncState EmptySyncState => new(new Dictionary()); + + [Fact] + public async Task ReadingIgnoresTruncatedLastLine() + { + var dir = Directory.CreateTempSubdirectory("harmony-json-syncable-test-"); + try + { + var syncable = new JsonSyncable(dir, SerializerOptions, NullLogger.Instance); + var clientId = Guid.NewGuid(); + Commit[] commits = [MakeCommit(clientId, 0), MakeCommit(clientId, 1)]; + await syncable.AddRangeFromSync(commits); + await File.AppendAllTextAsync(ClientFile(dir, clientId), Truncated(MakeCommit(clientId, 2)), + TestContext.Current.CancellationToken); + + var syncState = await syncable.GetSyncState(); + syncState.GetClientState(clientId)!.CommitCount.Should().Be(2); + + var changes = await syncable.GetChanges(EmptySyncState); + changes.MissingFromClient.Select(c => c.Id).Should().BeEquivalentTo(commits.Select(c => c.Id)); + } + finally + { + dir.Delete(recursive: true); + } + } + + [Theory] + [InlineData("\n")] + [InlineData("\r\n")] + public async Task AppendingRemovesTruncatedLastLine(string newline) + { + var dir = Directory.CreateTempSubdirectory("harmony-json-syncable-test-"); + try + { + var syncable = new JsonSyncable(dir, SerializerOptions, NullLogger.Instance); + var clientId = Guid.NewGuid(); + var file = ClientFile(dir, clientId); + var first = MakeCommit(clientId, 0); + var dropped = MakeCommit(clientId, 1); + await File.WriteAllTextAsync(file, Serialize(first) + newline + Truncated(dropped), + TestContext.Current.CancellationToken); + + var next = MakeCommit(clientId, 2); + await syncable.AddRangeFromSync([next]); + + var lines = await File.ReadAllLinesAsync(file, TestContext.Current.CancellationToken); + lines.Should().Equal(Serialize(first), Serialize(next)); + + //the device that wrote the dropped commit re-sends it on the next sync + await syncable.AddRangeFromSync([dropped]); + lines = await File.ReadAllLinesAsync(file, TestContext.Current.CancellationToken); + lines.Should().Equal(Serialize(first), Serialize(next), Serialize(dropped)); + + var changes = await syncable.GetChanges(EmptySyncState); + changes.MissingFromClient.Select(c => c.Id).Should() + .BeEquivalentTo([first.Id, next.Id, dropped.Id]); + } + finally + { + dir.Delete(recursive: true); + } + } + + [Fact] + public async Task AppendingRemovesFragmentWhenFileHasNoCompleteLines() + { + var dir = Directory.CreateTempSubdirectory("harmony-json-syncable-test-"); + try + { + var syncable = new JsonSyncable(dir, SerializerOptions, NullLogger.Instance); + var clientId = Guid.NewGuid(); + var file = ClientFile(dir, clientId); + await File.WriteAllTextAsync(file, Truncated(MakeCommit(clientId, 0)), TestContext.Current.CancellationToken); + + (await syncable.GetSyncState()).ClientStates.Should().BeEmpty(); + + var next = MakeCommit(clientId, 1); + await syncable.AddRangeFromSync([next]); + + var lines = await File.ReadAllLinesAsync(file, TestContext.Current.CancellationToken); + lines.Should().Equal(Serialize(next)); + } + finally + { + dir.Delete(recursive: true); + } + } + + [Theory] + [InlineData("")] + [InlineData("\r")] + public async Task UnterminatedCompleteLastLineIsKept(string trailing) + { + var dir = Directory.CreateTempSubdirectory("harmony-json-syncable-test-"); + try + { + var syncable = new JsonSyncable(dir, SerializerOptions, NullLogger.Instance); + var clientId = Guid.NewGuid(); + var file = ClientFile(dir, clientId); + var first = MakeCommit(clientId, 0); + var unterminated = MakeCommit(clientId, 1); + await syncable.AddRangeFromSync([first]); + await File.AppendAllTextAsync(file, Serialize(unterminated) + trailing, TestContext.Current.CancellationToken); + + (await syncable.GetSyncState()).GetClientState(clientId)!.CommitCount.Should().Be(2); + + var next = MakeCommit(clientId, 2); + await syncable.AddRangeFromSync([next]); + + var lines = await File.ReadAllLinesAsync(file, TestContext.Current.CancellationToken); + lines.Should().Equal(Serialize(first), Serialize(unterminated), Serialize(next)); + var changes = await syncable.GetChanges(EmptySyncState); + changes.MissingFromClient.Select(c => c.Id).Should() + .BeEquivalentTo([first.Id, unterminated.Id, next.Id]); + } + finally + { + dir.Delete(recursive: true); + } + } + + [Theory] + [InlineData("\n")] + [InlineData("\r\n")] + public async Task MalformedLineInTheMiddleThrows(string newline) + { + var dir = Directory.CreateTempSubdirectory("harmony-json-syncable-test-"); + try + { + var syncable = new JsonSyncable(dir, SerializerOptions, NullLogger.Instance); + var clientId = Guid.NewGuid(); + await File.WriteAllTextAsync(ClientFile(dir, clientId), + Serialize(MakeCommit(clientId, 0)) + newline + + Truncated(MakeCommit(clientId, 1)) + newline + + Serialize(MakeCommit(clientId, 2)) + newline, + TestContext.Current.CancellationToken); + + await syncable.Invoking(s => s.GetSyncState()).Should().ThrowAsync(); + await syncable.Invoking(s => s.GetChanges(EmptySyncState)).Should().ThrowAsync(); + await syncable.Invoking(s => s.AddRangeFromSync([MakeCommit(clientId, 3)])).Should().ThrowAsync(); + } + finally + { + dir.Delete(recursive: true); + } + } + + [Fact] + public async Task TerminatedMalformedLastLineThrows() + { + var dir = Directory.CreateTempSubdirectory("harmony-json-syncable-test-"); + try + { + var syncable = new JsonSyncable(dir, SerializerOptions, NullLogger.Instance); + var clientId = Guid.NewGuid(); + await File.WriteAllTextAsync(ClientFile(dir, clientId), + Serialize(MakeCommit(clientId, 0)) + "\n" + Truncated(MakeCommit(clientId, 1)) + "\n", + TestContext.Current.CancellationToken); + + await syncable.Invoking(s => s.GetSyncState()).Should().ThrowAsync(); + } + finally + { + dir.Delete(recursive: true); + } + } } diff --git a/src/SIL.Harmony/JsonSyncable.cs b/src/SIL.Harmony/JsonSyncable.cs index 73349b5e..9466286d 100644 --- a/src/SIL.Harmony/JsonSyncable.cs +++ b/src/SIL.Harmony/JsonSyncable.cs @@ -36,6 +36,7 @@ await Parallel.ForEachAsync(groups, async (group, ct) => using (await ClientLocks.GetOrAdd(group.Key, _ => new AsyncLock()).LockAsync()) { var file = FileForClientId(group.Key); + await RepairTrailingFragmentAsync(file, ct); var existingIds = await GetExistingCommitIdsAsync(file, ct); var newCommits = group.Where(c => existingIds.Add(c.Id)).DefaultOrder().ToArray(); if (newCommits.Length == 0) return; @@ -127,19 +128,128 @@ private static Guid ClientIdForFile(FileInfo clientIdFile) private async IAsyncEnumerable ReadAllCommitsAsync(FileInfo file, [EnumeratorCancellation] CancellationToken cancellationToken) { - if (!file.Exists || file.Length == 0) + if (!file.Exists) yield break; await using var stream = file.OpenRead(); + if (stream.Length == 0) + yield break; + var endsWithNewline = await EndsWithNewlineAsync(stream, cancellationToken); using var reader = new StreamReader(stream); - while (await reader.ReadLineAsync(cancellationToken) is { } line) + var line = await reader.ReadLineAsync(cancellationToken); + while (line is not null) { cancellationToken.ThrowIfCancellationRequested(); - if (string.IsNullOrWhiteSpace(line)) - continue; - var commit = JsonSerializer.Deserialize(line, _serializerOptions); - if (commit is not null) - yield return commit; + var next = await reader.ReadLineAsync(cancellationToken); + if (!string.IsNullOrWhiteSpace(line)) + { + Commit? commit; + //an unterminated last line is most likely a write that was interrupted (e.g. a drive pulled mid-sync), + //the commit still exists on the client that wrote it, so it will be re-sent on the next sync + if (next is null && !endsWithNewline) + { + if (!TryDeserializeCommit(line, out commit)) + { + _logger.LogWarning("Ignoring incomplete trailing line in {File}", file.FullName); + yield break; + } + } + else + { + commit = JsonSerializer.Deserialize(line, _serializerOptions); + } + + if (commit is not null) + yield return commit; + } + line = next; + } + } + + /// + /// Ensures the file ends with a newline so the next append starts on its own line. + /// A complete commit missing only its newline gets one; an incomplete fragment is truncated away. + /// Must be called while holding the client lock. + /// + private async Task RepairTrailingFragmentAsync(FileInfo file, CancellationToken cancellationToken) + { + if (!file.Exists) + return; + + await using var stream = new FileStream(file.FullName, FileMode.Open, FileAccess.ReadWrite, FileShare.Read); + var fragmentStart = await FindEndOfLastCompleteLineAsync(stream, cancellationToken); + if (fragmentStart == stream.Length) + return; + + var fragment = new byte[stream.Length - fragmentStart]; + stream.Position = fragmentStart; + await stream.ReadExactlyAsync(fragment, cancellationToken); + if (TryDeserializeCommit(fragment, out _)) + { + stream.Position = stream.Length; + await stream.WriteAsync("\n"u8.ToArray(), cancellationToken); + } + else + { + _logger.LogWarning("Truncating incomplete trailing line ({Length} bytes) from {File}", fragment.Length, file.FullName); + stream.SetLength(fragmentStart); + } + } + + private static async Task EndsWithNewlineAsync(FileStream stream, CancellationToken cancellationToken) + { + var lastByte = new byte[1]; + stream.Position = stream.Length - 1; + await stream.ReadExactlyAsync(lastByte, cancellationToken); + stream.Position = 0; + return lastByte[0] == (byte)'\n'; + } + + /// The position just past the last '\n' in the file, or 0 if there is none. + private static async Task FindEndOfLastCompleteLineAsync(FileStream stream, CancellationToken cancellationToken) + { + var buffer = new byte[4096]; + var end = stream.Length; + while (end > 0) + { + var start = Math.Max(0, end - buffer.Length); + var count = (int)(end - start); + stream.Position = start; + await stream.ReadExactlyAsync(buffer.AsMemory(0, count), cancellationToken); + //'\n' never occurs inside a multi-byte UTF-8 sequence, so a byte search is safe + var index = buffer.AsSpan(0, count).LastIndexOf((byte)'\n'); + if (index >= 0) + return start + index + 1; + end = start; + } + return 0; + } + + private bool TryDeserializeCommit(string json, out Commit? commit) + { + try + { + commit = JsonSerializer.Deserialize(json, _serializerOptions); + return true; + } + catch (JsonException) + { + commit = null; + return false; + } + } + + private bool TryDeserializeCommit(ReadOnlySpan utf8Json, out Commit? commit) + { + try + { + commit = JsonSerializer.Deserialize(utf8Json, _serializerOptions); + return true; + } + catch (JsonException) + { + commit = null; + return false; } }