Make JsonSyncable tolerate an interrupted write

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<IChange>
+            {
+                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<Guid, long>());
+
+    [Fact]
+    public async Task ReadingIgnoresTruncatedLastLine()
+    {
+        var dir = Directory.CreateTempSubdirectory("harmony-json-syncable-test-");
+        try
+        {
+            var syncable = new JsonSyncable(dir, SerializerOptions, NullLogger<JsonSyncable>.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<JsonSyncable>.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<JsonSyncable>.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<JsonSyncable>.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<JsonSyncable>.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<JsonException>();
+            await syncable.Invoking(s => s.GetChanges(EmptySyncState)).Should().ThrowAsync<JsonException>();
+            await syncable.Invoking(s => s.AddRangeFromSync([MakeCommit(clientId, 3)])).Should().ThrowAsync<JsonException>();
+        }
+        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<JsonSyncable>.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<JsonException>();
+        }
+        finally
+        {
+            dir.Delete(recursive: true);
+        }
+    }
 }

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<Commit> 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<Commit>(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<Commit>(line, _serializerOptions);
+                }
+
+                if (commit is not null)
+                    yield return commit;
+            }
+            line = next;
+        }
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    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<bool> 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';
+    }
+
+    /// <returns>The position just past the last '\n' in the file, or 0 if there is none.</returns>
+    private static async Task<long> 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<Commit>(json, _serializerOptions);
+            return true;
+        }
+        catch (JsonException)
+        {
+            commit = null;
+            return false;
+        }
+    }
+
+    private bool TryDeserializeCommit(ReadOnlySpan<byte> utf8Json, out Commit? commit)
+    {
+        try
+        {
+            commit = JsonSerializer.Deserialize<Commit>(utf8Json, _serializerOptions);
+            return true;
+        }
+        catch (JsonException)
+        {
+            commit = null;
+            return false;
         }
     }