Align all membership providers: ZooKeeper and Cosmos cleanup non-Active stale entries

- CosmosMembershipTable: remove all non-Active stale entries (was only Dead),
  using proper Math.Max(IAmAliveTime, StartTime) timestamp check
- ZooKeeperBasedMembershipTable: implement CleanupDefunctSiloEntries (was
  NotImplementedException) to delete non-Active stale entries

Co-authored-by: rkargMsft <164392675+rkargMsft@users.noreply.github.com>
Agent-Logs-Url: https://github.com/dotnet/orleans/sessions/cfa9474b-c8f1-4707-b13e-8ac1054679e7

Author: Copilot

src/Azure/Orleans.Clustering.Cosmos/Membership/CosmosMembershipTable.cs

@@ -103,7 +103,10 @@ public async Task CleanupDefunctSiloEntries(DateTimeOffset beforeDate)
     {
         try
         {
-            var silos = (await ReadSilos(SiloStatus.Dead).ConfigureAwait(false)).Where(s => s.IAmAliveTime < beforeDate).ToList();
+            var silos = (await ReadSilos().ConfigureAwait(false))
+                .Where(s => (SiloStatus)s.Status != SiloStatus.Active
+                    && new DateTime(Math.Max(s.IAmAliveTime.Ticks, s.StartTime.Ticks), DateTimeKind.Utc) < beforeDate)
+                .ToList();
             if (silos.Count == 0)
             {
                 return;

src/Orleans.Clustering.ZooKeeper/ZooKeeperBasedMembershipTable.cs

@@ -335,7 +335,23 @@ private static T Deserialize<T>(byte[] data)
 
         public Task CleanupDefunctSiloEntries(DateTimeOffset beforeDate)
         {
-            throw new NotImplementedException();
+            return UsingZookeeper(async zk =>
+            {
+                var childrenResult = await zk.getChildrenAsync("/");
+                var rows = await Task.WhenAll(
+                    childrenResult.Children.Select(child => GetRow(zk, SiloAddress.FromParsableString(child))));
+
+                foreach (var (entry, _) in rows)
+                {
+                    if (entry.Status != SiloStatus.Active
+                        && new DateTime(Math.Max(entry.IAmAliveTime.Ticks, entry.StartTime.Ticks), DateTimeKind.Utc) < beforeDate)
+                    {
+                        await ZKUtil.deleteRecursiveAsync(zk, ConvertToRowPath(entry.SiloAddress));
+                    }
+                }
+
+                return true;
+            }, this.deploymentConnectionString, this.watcher);
         }
 
         [LoggerMessage(