Add flow for re-sending replay frames that spectator did not receive at end of play

osu.Game.Tests/Visual/Multiplayer/MultiplayerGameplayLeaderboardTestScene.cs

@@ -246,7 +246,7 @@ protected void UpdateUserStatesRandomly()
                         break;
                 }
 
-                spectatorClient.Raise(s => s.OnNewFrames -= null, userId, new FrameDataBundle(header, new[] { new LegacyReplayFrame(Time.Current, 0, 0, ReplayButtonState.None) }));
+                spectatorClient.Raise(s => s.OnNewFrames -= null, userId, new FrameDataBundle(header, new[] { new LegacyReplayFrame(Time.Current, 0, 0, ReplayButtonState.None) }, null));
             }
         }
     }

osu.Game/Online/Spectator/CompleteReplayRequest.cs

@@ -0,0 +1,21 @@
+// Copyright (c) ppy Pty Ltd <contact@ppy.sh>. Licensed under the MIT Licence.
+// See the LICENCE file in the repository root for full licence text.
+
+using System;
+using System.Collections.Generic;
+using MessagePack;
+
+namespace osu.Game.Online.Spectator
+{
+    /// <summary>
+    /// Sent by the server to retrieve frame bundles that the client sent but never reached the server, if any.
+    /// </summary>
+    /// <param name="ScoreTokenId">The ID of the score token associated with the score with missing bundles.</param>
+    /// <param name="FrameBundleSequenceNumbers">The sequence numbers of frame bundles that the server never received.</param>
+    [Serializable]
+    [MessagePackObject]
+    public record CompleteReplayRequest(
+        [property: Key(0)] long ScoreTokenId,
+        [property: Key(1)] IEnumerable<long> FrameBundleSequenceNumbers
+    );
+}

osu.Game/Online/Spectator/CompleteReplayResponse.cs

@@ -0,0 +1,22 @@
+// Copyright (c) ppy Pty Ltd <contact@ppy.sh>. Licensed under the MIT Licence.
+// See the LICENCE file in the repository root for full licence text.
+
+using System;
+using System.Collections.Generic;
+using MessagePack;
+
+namespace osu.Game.Online.Spectator
+{
+    /// <summary>
+    /// Response to <see cref="CompleteReplayRequest"/>.
+    /// </summary>
+    /// <param name="FrameBundles">
+    /// The frame bundles requested by sequence number in the corresponding <see cref="CompleteReplayRequest"/>.
+    /// Can be blank (contain no frames) if the client does not have the frames available any more.
+    /// </param>
+    [Serializable]
+    [MessagePackObject]
+    public record CompleteReplayResponse(
+        [property: Key(0)] IEnumerable<FrameDataBundle> FrameBundles
+    );
+}

osu.Game/Online/Spectator/FrameDataBundle.cs

@@ -21,17 +21,25 @@ public class FrameDataBundle
         [Key(1)]
         public IList<LegacyReplayFrame> Frames { get; set; }
 
+        /// <summary>
+        /// The sequence number of this frame bundle.
+        /// Used to determine ordering of frame bundles, and for server-side checks that server received all frame bundles it was supposed to.
+        /// </summary>
+        [Key(2)]
+        public long? SequenceNumber { get; set; }
+
         public FrameDataBundle(ScoreInfo score, ScoreProcessor scoreProcessor, IList<LegacyReplayFrame> frames)
         {
             Frames = frames;
             Header = new FrameHeader(score, scoreProcessor.GetScoreProcessorStatistics());
         }
 
         [JsonConstructor]
-        public FrameDataBundle(FrameHeader header, IList<LegacyReplayFrame> frames)
+        public FrameDataBundle(FrameHeader header, IList<LegacyReplayFrame> frames, long? sequenceNumber)
         {
             Header = header;
             Frames = frames;
+            SequenceNumber = sequenceNumber;
         }
     }
 }

osu.Game/Online/Spectator/ISpectatorClient.cs

@@ -49,5 +49,12 @@ public interface ISpectatorClient : IStatefulUserHubClient
         /// </summary>
         /// <param name="userId">The ID of the user who ended watching.</param>
         Task UserEndedWatching(int userId);
+
+        /// <summary>
+        /// Called by the server in response to <see cref="ISpectatorServer.EndPlaySession"/>.
+        /// Using data provided in <see cref="SpectatorState"/>, this gives the server a last chance to retrieve any <see cref="FrameDataBundle"/>s
+        /// that it may not have received due to connection drop-outs or similar.
+        /// </summary>
+        Task<CompleteReplayResponse> CompleteReplay(CompleteReplayRequest request);
     }
 }

osu.Game/Online/Spectator/OnlineSpectatorClient.cs

@@ -42,6 +42,7 @@ private void load(IAPIProvider api)
                     connection.On<int, SpectatorState>(nameof(ISpectatorClient.UserBeganPlaying), ((ISpectatorClient)this).UserBeganPlaying);
                     connection.On<int, FrameDataBundle>(nameof(ISpectatorClient.UserSentFrames), ((ISpectatorClient)this).UserSentFrames);
                     connection.On<int, SpectatorState>(nameof(ISpectatorClient.UserFinishedPlaying), ((ISpectatorClient)this).UserFinishedPlaying);
+                    connection.On<CompleteReplayRequest, CompleteReplayResponse>(nameof(ISpectatorClient.CompleteReplay), ((ISpectatorClient)this).CompleteReplay);
                     connection.On<int, long>(nameof(ISpectatorClient.UserScoreProcessed), ((ISpectatorClient)this).UserScoreProcessed);
                     connection.On<SpectatorUser[]>(nameof(ISpectatorClient.UserStartedWatching), ((ISpectatorClient)this).UserStartedWatching);
                     connection.On<int>(nameof(ISpectatorClient.UserEndedWatching), ((ISpectatorClient)this).UserEndedWatching);

osu.Game/Online/Spectator/SpectatorClient.cs

@@ -5,6 +5,7 @@
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using JetBrains.Annotations;
 using osu.Framework.Allocation;
@@ -89,8 +90,10 @@ public abstract partial class SpectatorClient : Component, ISpectatorClient
         private Score? currentScore;
         private long? currentScoreToken;
         private ScoreProcessor? currentScoreProcessor;
+        private long currentFrameBundleSequenceNumber;
 
         private readonly Queue<FrameDataBundle> pendingFrameBundles = new Queue<FrameDataBundle>();
+        private readonly Dictionary<long, List<FrameDataBundle>> allFrameBundles = new Dictionary<long, List<FrameDataBundle>>();
 
         private readonly List<LegacyReplayFrame> pendingFrames = new List<LegacyReplayFrame>();
 
@@ -284,6 +287,8 @@ public void EndPlaying(GameplayState state)
                 if (pendingFrames.Count > 0)
                     purgePendingFrames();
 
+                currentState.LastFrameBundleSequenceNumber = currentFrameBundleSequenceNumber;
+
                 clearScoreState();
 
                 if (state.HasPassed)
@@ -305,6 +310,9 @@ private void setStateForScore(long? scoreToken, GameplayState state, Score score
             currentScore = score;
             currentScoreToken = scoreToken;
             currentScoreProcessor = state.ScoreProcessor;
+            currentFrameBundleSequenceNumber = 0;
+            if (scoreToken != null)
+                allFrameBundles[scoreToken.Value] = new List<FrameDataBundle>();
         }
 
         private void clearScoreState()
@@ -315,6 +323,22 @@ private void clearScoreState()
             currentScore = null;
             currentScoreProcessor = null;
             currentScoreToken = null;
+            currentFrameBundleSequenceNumber = 0;
+        }
+
+        public Task<CompleteReplayResponse> CompleteReplay(CompleteReplayRequest completeReplayRequest)
+        {
+            if (!allFrameBundles.Remove(completeReplayRequest.ScoreTokenId, out var frameBundlesForScoreToken))
+                return Task.FromResult(new CompleteReplayResponse([]));
+
+            var sequenceNumberSet = completeReplayRequest.FrameBundleSequenceNumbers.ToHashSet();
+            if (sequenceNumberSet.Count == 0)
+                return Task.FromResult(new CompleteReplayResponse([]));
+
+            var bundles = frameBundlesForScoreToken.Where(b => b.SequenceNumber != null && sequenceNumberSet.Contains(b.SequenceNumber.Value))
+                                                   .OrderBy(b => b.SequenceNumber)
+                                                   .ToArray();
+            return Task.FromResult(new CompleteReplayResponse(bundles));
         }
 
         public virtual void WatchUser(int userId)
@@ -389,7 +413,16 @@ private void purgePendingFrames()
             Debug.Assert(currentScoreProcessor != null);
 
             var frames = pendingFrames.ToArray();
-            var bundle = new FrameDataBundle(currentScore.ScoreInfo, currentScoreProcessor, frames);
+            var bundle = new FrameDataBundle(currentScore.ScoreInfo, currentScoreProcessor, frames)
+            {
+                SequenceNumber = Interlocked.Increment(ref currentFrameBundleSequenceNumber)
+            };
+
+            if (currentScoreToken != null)
+            {
+                Debug.Assert(allFrameBundles.ContainsKey(currentScoreToken.Value));
+                allFrameBundles[currentScoreToken.Value].Add(bundle);
+            }
 
             pendingFrames.Clear();
             lastPurgeTime = Time.Current;

osu.Game/Online/Spectator/SpectatorState.cs

@@ -33,6 +33,13 @@ public class SpectatorState : IEquatable<SpectatorState>
         [Key(4)]
         public Dictionary<HitResult, int> MaximumStatistics { get; set; } = new Dictionary<HitResult, int>();
 
+        /// <summary>
+        /// The sequence number of the last frame bundle in the replay.
+        /// Only available in <see cref="ISpectatorServer.EndPlaySession"/> invocations.
+        /// </summary>
+        [Key(5)]
+        public long? LastFrameBundleSequenceNumber { get; set; }
+
         public bool Equals(SpectatorState other)
         {
             if (ReferenceEquals(null, other)) return false;