Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
10 changes: 10 additions & 0 deletions src/SIL.Harmony.Core/HarmonyActivitySource.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,10 @@
using System.Diagnostics;

namespace SIL.Harmony.Core;

public static class HarmonyActivitySource
{
public const string ActivitySourceName = "SIL.Harmony";
internal static readonly ActivitySource Source =
new(ActivitySourceName, typeof(HarmonyActivitySource).Assembly.GetName().Version?.ToString());
}
59 changes: 53 additions & 6 deletions src/SIL.Harmony.Core/QueryHelpers.cs
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
using System.Diagnostics;
using System.IO.Hashing;
using Microsoft.EntityFrameworkCore;

Expand Down Expand Up @@ -46,11 +47,10 @@ public static async IAsyncEnumerable<TCommit> GetCommitsMissingFromRemote<TCommi
{
commits = commits.AsNoTracking();
if (includeChangeEntities) commits = commits.Include(c => c.ChangeEntities);
foreach (var localClientState in localState.ClientStates)
foreach (var (localClientState, remoteClientState, plan) in PlanClients(localState, remoteState))
{
var clientCommits = commits.Where(c => c.ClientId == localClientState.ClientId).DefaultOrder();
var remoteClientState = remoteState.GetClientState(localClientState.ClientId);
switch (PlanFor(localClientState, remoteClientState))
switch (plan)
{
case SyncPlan.SendNothing:
break;
Expand Down Expand Up @@ -98,6 +98,54 @@ private static SyncPlan PlanFor(ClientState local, ClientState? remote)
return local.CommitCount >= remote.CommitCount ? SyncPlan.SendAll : SyncPlan.SendNothing;
}

//planning only reads the two states, so the span ends before either iterator yields and Activity.Current never crosses a yield
private static (ClientState Local, ClientState? Remote, SyncPlan Plan)[] PlanClients(SyncState localState, SyncState remoteState)
{
using var activity = StartPlanActivity(remoteState);
return [.. localState.ClientStates.Select(local =>
{
var remote = remoteState.GetClientState(local.ClientId);
return (local, remote, PlanAndRecordRecovery(activity, local, remote));
})];
}

private static Activity? StartPlanActivity(SyncState remoteState)
{
var activity = HarmonyActivitySource.Source.StartActivity(nameof(PlanClients));
if (activity is { IsAllDataRequested: true } && remoteState.ClientStates.Any(c => c.OnlyHasTimestamp))
activity.SetTag("harmony.sync.remote_heads_only", true);
return activity;
}

private static SyncPlan PlanAndRecordRecovery(Activity? activity, ClientState local, ClientState? remote)
{
var plan = PlanFor(local, remote);
if (activity is { IsAllDataRequested: true } && remote is { OnlyHasTimestamp: false } && IsRecovering(local, remote))
RecordRecovery(activity, local, remote, plan);
return plan;
}

//a side sends its whole history for this client only when PlanFor falls back to recovery
private static bool IsRecovering(ClientState local, ClientState remote) =>
local != remote
&& (PlanFor(local, remote) == SyncPlan.SendAll || PlanFor(remote, local) == SyncPlan.SendAll);

private static void RecordRecovery(Activity activity, ClientState local, ClientState remote, SyncPlan plan)
{
activity.AddEvent(new ActivityEvent("harmony.sync.recovery", tags: new ActivityTagsCollection
{
["harmony.client_id"] = local.ClientId.ToString(),
["harmony.plan"] = plan.ToString(),
["harmony.local.commit_count"] = local.CommitCount,
["harmony.local.max_timestamp"] = local.MaxTimestamp,
//OTLP has no unsigned 64-bit attribute type
["harmony.local.hash"] = local.Hash.ToString("x16"),
["harmony.remote.commit_count"] = remote.CommitCount,
["harmony.remote.max_timestamp"] = remote.MaxTimestamp,
["harmony.remote.hash"] = remote.Hash.ToString("x16")
}));
}

//the db keeps sub-millisecond precision, so re-check against the millisecond value the remote reported
private static bool IsAfter(CommitBase commit, long afterMs) =>
commit.DateTime.ToUnixTimeMilliseconds() > afterMs;
Expand All @@ -122,11 +170,10 @@ public static IEnumerable<TCommit> GetCommitsMissingFromRemote<TCommit, TChange>
SyncState localState,
SyncState remoteState) where TCommit : CommitBase<TChange>
{
foreach (var localClientState in localState.ClientStates)
foreach (var (localClientState, remoteClientState, plan) in PlanClients(localState, remoteState))
{
ClientState? remoteClientState = remoteState.GetClientState(localClientState.ClientId);
var clientCommits = commits.Where(c => c.ClientId == localClientState.ClientId).DefaultOrder();
switch (PlanFor(localClientState, remoteClientState))
switch (plan)
{
case SyncPlan.SendNothing:
break;
Expand Down
Loading