diff --git a/src/Tgstation.Server.Host/Controllers/SwarmController.cs b/src/Tgstation.Server.Host/Controllers/SwarmController.cs index 4ef0f74be1..0bbc78686c 100644 --- a/src/Tgstation.Server.Host/Controllers/SwarmController.cs +++ b/src/Tgstation.Server.Host/Controllers/SwarmController.cs @@ -86,9 +86,10 @@ namespace Tgstation.Server.Host.Controllers /// Registration endpoint. /// /// The . + /// The for the operation. /// The of the operation. [HttpPost(SwarmConstants.RegisterRoute)] - public IActionResult Register([FromBody] SwarmRegistrationRequest registrationRequest) + public async Task Register([FromBody] SwarmRegistrationRequest registrationRequest, CancellationToken cancellationToken) { if (registrationRequest == null) throw new ArgumentNullException(nameof(registrationRequest)); @@ -96,7 +97,7 @@ namespace Tgstation.Server.Host.Controllers if (registrationRequest.ServerVersion != assemblyInformationProvider.Version) return StatusCode((int)HttpStatusCode.UpgradeRequired); - var registrationResult = swarmOperations.RegisterNode(registrationRequest, RequestRegistrationId); + var registrationResult = await swarmOperations.RegisterNode(registrationRequest, RequestRegistrationId, cancellationToken); if (!registrationResult) return Conflict(); return NoContent(); @@ -201,15 +202,14 @@ namespace Tgstation.Server.Host.Controllers /// /// Update abort endpoint. /// - /// The for the operation. /// A resulting in the of the operation. [HttpDelete(SwarmConstants.UpdateRoute)] - public async Task AbortUpdate(CancellationToken cancellationToken) + public async Task AbortUpdate() { if (!ValidateRegistration()) return Forbid(); - await swarmOperations.RemoteAbortUpdate(cancellationToken); + await swarmOperations.AbortUpdate(); return NoContent(); } diff --git a/src/Tgstation.Server.Host/Core/ServerUpdater.cs b/src/Tgstation.Server.Host/Core/ServerUpdater.cs index 529a8a3a69..b1c0b10c53 100644 --- a/src/Tgstation.Server.Host/Core/ServerUpdater.cs +++ b/src/Tgstation.Server.Host/Core/ServerUpdater.cs @@ -139,9 +139,9 @@ namespace Tgstation.Server.Host.Core if (!tuple.Item2) await bufferedStream.DisposeAsync(); // don't leave this in memory } - catch (Exception ex) when (ex is not OperationCanceledException) + catch (Exception ex) { - await TryAbort(ex, cancellationToken); + await TryAbort(ex); throw; } @@ -193,15 +193,14 @@ namespace Tgstation.Server.Host.Core /// Attempt to abort a prepared swarm update. /// /// The being thrown. - /// The for the operation. /// A representing the running operation. /// A new containing and the swarm abort if thrown. /// Requires to be populated. - async Task TryAbort(Exception exception, CancellationToken cancellationToken) + async Task TryAbort(Exception exception) { try { - await serverUpdateOperation.SwarmService.AbortUpdate(cancellationToken); + await serverUpdateOperation.SwarmService.AbortUpdate(); } catch (Exception e2) { @@ -251,7 +250,7 @@ namespace Tgstation.Server.Host.Core } catch (Exception ex) { - await TryAbort(ex, cancellationToken); + await TryAbort(ex); throw; } } diff --git a/src/Tgstation.Server.Host/Swarm/ISwarmOperations.cs b/src/Tgstation.Server.Host/Swarm/ISwarmOperations.cs index 77054861db..a2cae0c47c 100644 --- a/src/Tgstation.Server.Host/Swarm/ISwarmOperations.cs +++ b/src/Tgstation.Server.Host/Swarm/ISwarmOperations.cs @@ -10,7 +10,7 @@ namespace Tgstation.Server.Host.Swarm /// /// Swarm service operations for the . /// - public interface ISwarmOperations + public interface ISwarmOperations : ISwarmUpdateAborter { /// /// Pass in an updated list of to the node. @@ -38,8 +38,9 @@ namespace Tgstation.Server.Host.Swarm /// /// The that is registering. /// The registration . - /// if the registration was successful, otherwise. - bool RegisterNode(Api.Models.Internal.SwarmServer node, Guid registrationId); + /// The for the operation. + /// A resulting in if the registration was successful, otherwise. + Task RegisterNode(Api.Models.Internal.SwarmServer node, Guid registrationId, CancellationToken cancellationToken); /// /// Attempt to unregister a node with a given with the controller. @@ -56,12 +57,5 @@ namespace Tgstation.Server.Host.Swarm /// The for the operation. /// A representing the running operation. Task RemoteCommitRecieved(Guid registrationId, CancellationToken cancellationToken); - - /// - /// Remotely abort an uncommitted update. - /// - /// The for the operation. - /// A representing the running operation. - Task RemoteAbortUpdate(CancellationToken cancellationToken); } } diff --git a/src/Tgstation.Server.Host/Swarm/ISwarmService.cs b/src/Tgstation.Server.Host/Swarm/ISwarmService.cs index f366740da3..c8d3866320 100644 --- a/src/Tgstation.Server.Host/Swarm/ISwarmService.cs +++ b/src/Tgstation.Server.Host/Swarm/ISwarmService.cs @@ -11,7 +11,7 @@ namespace Tgstation.Server.Host.Swarm /// /// Used for swarm operations. Functions may be no-op based on configuration. /// - public interface ISwarmService + public interface ISwarmService : ISwarmUpdateAborter { /// /// Gets a value indicating if the expected amount of nodes are connected to the swarm. @@ -39,12 +39,5 @@ namespace Tgstation.Server.Host.Swarm /// /// A of s in the swarm. If the server is not part of a swarm, will be returned. ICollection GetSwarmServers(); - - /// - /// Abort an uncommitted update. - /// - /// The for the operation. - /// A representing the running operation. - Task AbortUpdate(CancellationToken cancellationToken); } } diff --git a/src/Tgstation.Server.Host/Swarm/ISwarmUpdateAborter.cs b/src/Tgstation.Server.Host/Swarm/ISwarmUpdateAborter.cs new file mode 100644 index 0000000000..e3f59893f6 --- /dev/null +++ b/src/Tgstation.Server.Host/Swarm/ISwarmUpdateAborter.cs @@ -0,0 +1,17 @@ +using System.Threading.Tasks; + +namespace Tgstation.Server.Host.Swarm +{ + /// + /// Allows aborting a swarm distributed update operation. + /// + public interface ISwarmUpdateAborter + { + /// + /// Attempt to abort an uncommitted update. + /// + /// A representing the running operation. + /// This method does not accept a because aborting an update should never be cancelled. + Task AbortUpdate(); + } +} diff --git a/src/Tgstation.Server.Host/Swarm/SwarmService.cs b/src/Tgstation.Server.Host/Swarm/SwarmService.cs index 0d29880ce5..fc4272b959 100644 --- a/src/Tgstation.Server.Host/Swarm/SwarmService.cs +++ b/src/Tgstation.Server.Host/Swarm/SwarmService.cs @@ -12,6 +12,7 @@ using System.Web; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; + using Newtonsoft.Json; using Newtonsoft.Json.Serialization; @@ -137,41 +138,26 @@ namespace Tgstation.Server.Host.Swarm /// readonly Dictionary registrationIds; - /// - /// used for accessing . - /// - readonly object updateSynchronizationLock; - /// /// If the current server is the swarm controller. /// readonly bool swarmController; + /// + /// A that is currently in progress. + /// + volatile SwarmUpdateOperation updateOperation; + /// /// A that is used to force a health check. /// - TaskCompletionSource forceHealthCheckTcs; - - /// - /// The that is used to proceed with committing an update. - /// - TaskCompletionSource updateCommitTcs; - - /// - /// of s that need to send a ready-commit before the update can proceed. - /// - List nodesThatNeedToBeReadyToCommit; + volatile TaskCompletionSource forceHealthCheckTcs; /// /// The for the . /// Task serverHealthCheckTask; - /// - /// The set for a two phase commit update. - /// - Version targetUpdateVersion; - /// /// The registration provided by the swarm controller. /// @@ -187,11 +173,6 @@ namespace Tgstation.Server.Host.Swarm /// bool serversDirty; - /// - /// If we've sent out a remote commit message for an update operation. - /// - bool updateCommitSent; - /// /// Initializes static members of the class. /// @@ -271,8 +252,6 @@ namespace Tgstation.Server.Host.Swarm Identifier = swarmConfiguration.Identifier, }, }; - - updateSynchronizationLock = new object(); } } @@ -280,71 +259,31 @@ namespace Tgstation.Server.Host.Swarm public void Dispose() => serverHealthCheckCancellationTokenSource?.Dispose(); /// - public async Task RemoteAbortUpdate(CancellationToken cancellationToken) - { - if (targetUpdateVersion == null) - { - logger.LogTrace("Not remote aborting non-existent update"); - return; - } - - await AbortUpdate(cancellationToken); - } - - /// - public async Task AbortUpdate(CancellationToken cancellationToken) + public async Task AbortUpdate() { if (!SwarmMode) return; - logger.LogInformation("Aborting swarm update!"); - var commitTcs = updateCommitTcs; - updateCommitTcs = null; - commitTcs?.TrySetResult(false); - nodesThatNeedToBeReadyToCommit = null; - targetUpdateVersion = null; - - using var httpClient = httpClientFactory.CreateClient(); - async Task SendRemoteAbort(SwarmServerResponse swarmServer) + var localUpdateOperation = Interlocked.Exchange(ref updateOperation, null); + var abortResult = localUpdateOperation?.Abort(); + switch (abortResult) { - using var request = PrepareSwarmRequest( - swarmServer, - HttpMethod.Delete, - SwarmConstants.UpdateRoute, - null); - - try - { - using var response = await httpClient.SendAsync(request, HttpCompletionOption.ResponseContentRead, cancellationToken); - response.EnsureSuccessStatusCode(); - } - catch (Exception ex) - { - logger.LogWarning( - ex, - "Unable to set remote abort to {nodeOrController}!", - swarmController - ? $"node {swarmServer.Identifier}" - : "controller"); - } + case SwarmUpdateAbortResult.Aborted: + break; + case SwarmUpdateAbortResult.AlreadyAborted: + logger.LogDebug("Another context already aborted this update."); + return; + case SwarmUpdateAbortResult.CantAbortCommitted: + logger.LogDebug("Not aborting update because we have committed!"); + return; + case null: + logger.LogTrace("Attempted update abort but no operation was found!"); + return; + default: + throw new InvalidOperationException($"Invalid return value for SwarmUpdateOperation.Abort(): {abortResult}"); } - Task task; - if (!swarmController) - task = SendRemoteAbort(new SwarmServerResponse - { - Address = swarmConfiguration.ControllerAddress, - }); - else - { - lock (swarmServers) - task = Task.WhenAll( - swarmServers - .Where(x => !x.Controller) - .Select(SendRemoteAbort)); - } - - await task; + await RemoteAbortUpdate(); } /// @@ -354,11 +293,11 @@ namespace Tgstation.Server.Host.Swarm return SwarmCommitResult.ContinueUpdateNonCommitted; // wait for the update commit TCS - var commitTcsTask = updateCommitTcs?.Task; - if (commitTcsTask == null) + var localUpdateOperation = updateOperation; + if (localUpdateOperation == null) { - logger.LogDebug("Update commit failed, no pending task completion source!"); - await AbortUpdate(cancellationToken); + logger.LogDebug("Update commit failed, no pending operation!"); + await AbortUpdate(); // unnecessary, but can never be too safe return SwarmCommitResult.AbortUpdate; } @@ -382,7 +321,7 @@ namespace Tgstation.Server.Host.Swarm catch (Exception ex) { logger.LogWarning(ex, "Unable to send ready-commit to swarm controller!"); - await AbortUpdate(cancellationToken); + await AbortUpdate(); return SwarmCommitResult.AbortUpdate; } } @@ -393,13 +332,13 @@ namespace Tgstation.Server.Host.Swarm cancellationToken) : Extensions.TaskExtensions.InfiniteTask.WithToken(cancellationToken); - var commitTask = Task.WhenAny(commitTcsTask, timeoutTask); + var commitTask = Task.WhenAny(localUpdateOperation.CommitGate, timeoutTask); await commitTask; - var commitGoAhead = commitTcsTask.IsCompleted - && commitTcsTask.Result - && updateCommitTcs?.Task == commitTcsTask; + var commitGoAhead = localUpdateOperation.CommitGate.IsCompleted + && localUpdateOperation.CommitGate.Result + && localUpdateOperation == updateOperation; if (!commitGoAhead) { logger.LogDebug( @@ -407,7 +346,7 @@ namespace Tgstation.Server.Host.Swarm timeoutTask.IsCompleted ? " Timed out!" : String.Empty); - await AbortUpdate(cancellationToken); + await AbortUpdate(); return SwarmCommitResult.AbortUpdate; } @@ -441,8 +380,6 @@ namespace Tgstation.Server.Host.Swarm } } - updateCommitSent = true; - Task task; lock (swarmServers) task = Task.WhenAll( @@ -578,16 +515,18 @@ namespace Tgstation.Server.Host.Swarm return; } + // We're in a single-threaded-like context now so touching updateOperation directly is fine + // downgrade the db if necessary - if (targetUpdateVersion != null - && targetUpdateVersion < assemblyInformationProvider.Version) + if (updateOperation != null + && updateOperation.TargetVersion < assemblyInformationProvider.Version) await databaseContextFactory.UseContext( - db => databaseSeeder.Downgrade(db, targetUpdateVersion, cancellationToken)); + db => databaseSeeder.Downgrade(db, updateOperation.TargetVersion, cancellationToken)); if (SwarmMode) { // Put the nodes into a reconnecting state - if (targetUpdateVersion == null) + if (updateOperation == null) { logger.LogInformation("Unregistering nodes..."); Task task; @@ -640,7 +579,7 @@ namespace Tgstation.Server.Host.Swarm } /// - public bool RegisterNode(Api.Models.Internal.SwarmServer node, Guid registrationId) + public async Task RegisterNode(Api.Models.Internal.SwarmServer node, Guid registrationId, CancellationToken cancellationToken) { if (node == null) throw new ArgumentNullException(nameof(node)); @@ -654,48 +593,43 @@ namespace Tgstation.Server.Host.Swarm if (!swarmController) throw new InvalidOperationException("Cannot RegisterNode on swarm node!"); - lock (updateSynchronizationLock) + logger.LogTrace("RegisterNode"); + + await AbortUpdate(); + + lock (swarmServers) { - if (targetUpdateVersion != null) + if (registrationIds.Any(x => x.Value == registrationId)) { - logger.LogInformation("Not registering node {nodeId} as a distributed update is in progress.", node.Identifier); + var preExistingRegistrationKvp = registrationIds.FirstOrDefault(x => x.Value == registrationId); + if (preExistingRegistrationKvp.Key == node.Identifier) + { + logger.LogWarning("Node {nodeId} has already registered!", node.Identifier); + return true; + } + + logger.LogWarning( + "Registration ID collision! Node {nodeId} tried to register with {otherNodeId}'s registration ID: {registrationId}", + node.Identifier, + preExistingRegistrationKvp.Key, + registrationId); return false; } - lock (swarmServers) + if (registrationIds.TryGetValue(node.Identifier, out var oldRegistration)) { - if (registrationIds.Any(x => x.Value == registrationId)) - { - var preExistingRegistrationKvp = registrationIds.FirstOrDefault(x => x.Value == registrationId); - if (preExistingRegistrationKvp.Key == node.Identifier) - { - logger.LogWarning("Node {nodeId} has already registered!", node.Identifier); - return true; - } - - logger.LogWarning( - "Registration ID collision! Node {nodeId} tried to register with {otherNodeId}'s registration ID: {registrationId}", - node.Identifier, - preExistingRegistrationKvp.Key, - registrationId); - return false; - } - - if (registrationIds.TryGetValue(node.Identifier, out var oldRegistration)) - { - logger.LogInformation("Node {nodeId} is re-registering without first unregistering. Indicative of restart.", node.Identifier); - swarmServers.RemoveAll(x => x.Identifier == node.Identifier); - registrationIds.Remove(node.Identifier); - } - - swarmServers.Add(new SwarmServerResponse - { - Address = node.Address, - Identifier = node.Identifier, - Controller = false, - }); - registrationIds.Add(node.Identifier, registrationId); + logger.LogInformation("Node {nodeId} is re-registering without first unregistering. Indicative of restart.", node.Identifier); + swarmServers.RemoveAll(x => x.Identifier == node.Identifier); + registrationIds.Remove(node.Identifier); } + + swarmServers.Add(new SwarmServerResponse + { + Address = node.Address, + Identifier = node.Identifier, + Controller = false, + }); + registrationIds.Add(node.Identifier, registrationId); } logger.LogInformation("Registered node {nodeId} ({nodeIP}) with ID {registrationId}", node.Identifier, node.Address, registrationId); @@ -706,48 +640,58 @@ namespace Tgstation.Server.Host.Swarm /// public async Task RemoteCommitRecieved(Guid registrationId, CancellationToken cancellationToken) { + var localUpdateOperation = updateOperation; if (!swarmController) { logger.LogDebug("Received remote commit go ahead"); - var commitTcs = updateCommitTcs; - commitTcs?.TrySetResult(true); - return commitTcs != null; + return localUpdateOperation?.Commit() == true; } var nodeIdentifier = NodeIdentifierFromRegistration(registrationId); if (nodeIdentifier == null) { // Something fucky is happening, take no chances. - logger.LogDebug("Aborting update due to unforseen circumstances!"); - await AbortUpdate(cancellationToken); + logger.LogError("Aborting update due to unforseen circumstances!"); + await AbortUpdate(); return false; } - var nodeList = nodesThatNeedToBeReadyToCommit; - if (nodeList == null) + if (localUpdateOperation == null) { logger.LogDebug("Ignoring ready-commit from node {nodeId} as the update appears to have been aborted.", nodeIdentifier); return false; } - logger.LogDebug("Node {nodeId} is ready to commit.", nodeIdentifier); - lock (nodeList) + if (!localUpdateOperation.MarkNodeReady(nodeIdentifier)) { - nodeList.Remove(nodeIdentifier); - if (nodeList.Count == 0) + logger.LogError( + "Attempting to mark {nodeId} as ready to commit resulted in the update being aborted!", + nodeIdentifier); + + // bit racy here, localUpdateOperation has already been aborted. + // now if, FOR SOME GODFORSAKEN REASON, there's a new update operation, abort that too. + if (Interlocked.CompareExchange(ref updateOperation, null, localUpdateOperation) == localUpdateOperation) + await RemoteAbortUpdate(); + else { - logger.LogTrace("All nodes ready, update commit is a go once controller is ready"); - var commitTcs = updateCommitTcs; - return commitTcs?.TrySetResult(true) == true; + // marking as an error because how the actual fuck + logger.LogError("Aborting new update due to unforseen consequences!"); + await AbortUpdate(); } + + return false; } + logger.LogDebug("Node {nodeId} is ready to commit.", nodeIdentifier); return true; } /// public async Task UnregisterNode(Guid registrationId, CancellationToken cancellationToken) { + logger.LogTrace("UnregisterNode {registrationId}", registrationId); + await AbortUpdate(); + if (!swarmController) { // immediately trigger a health check @@ -757,18 +701,12 @@ namespace Tgstation.Server.Host.Swarm return; } - logger.LogTrace("UnregisterNode {registrationId}", registrationId); var nodeIdentifier = NodeIdentifierFromRegistration(registrationId); if (nodeIdentifier == null) return; logger.LogInformation("Unregistering node {nodeId}...", nodeIdentifier); - if (!updateCommitSent) - await AbortUpdate(cancellationToken); - else - logger.LogTrace("Not aborting update because we have committed"); - lock (swarmServers) { swarmServers.RemoveAll(x => x.Identifier == nodeIdentifier); @@ -778,6 +716,53 @@ namespace Tgstation.Server.Host.Swarm MarkServersDirty(); } + /// + /// Sends out remote abort update requests. + /// + /// A representing the running operation. + /// The aborted should be cleared out before calling this. This method does not accept a because aborting an update should never be cancelled. + Task RemoteAbortUpdate() + { + logger.LogInformation("Aborting swarm update!"); + + using var httpClient = httpClientFactory.CreateClient(); + async Task SendRemoteAbort(SwarmServerResponse swarmServer) + { + using var request = PrepareSwarmRequest( + swarmServer, + HttpMethod.Delete, + SwarmConstants.UpdateRoute, + null); + + try + { + // DCT: Intentionally should not be cancelled + using var response = await httpClient.SendAsync(request, HttpCompletionOption.ResponseContentRead, default); + response.EnsureSuccessStatusCode(); + } + catch (Exception ex) + { + logger.LogWarning( + ex, + "Unable to send remote abort to {nodeOrController}!", + swarmController + ? $"node {swarmServer.Identifier}" + : "controller"); + } + } + + if (!swarmController) + return SendRemoteAbort(new SwarmServerResponse + { + Address = swarmConfiguration.ControllerAddress, + }); + + return Task.WhenAll( + swarmServers + .Where(x => !x.Controller) + .Select(SendRemoteAbort)); + } + /// /// Create the for an update package retrieval from a given . /// @@ -817,44 +802,58 @@ namespace Tgstation.Server.Host.Swarm /// Implementation of ,. /// /// The containing the update package if this is the initiating server, otherwise. - /// The . + /// The . Must always have populated. If is , it must be fully populated. /// The for the operation. /// A resulting in the . async Task PrepareUpdateImpl(ISeekableFileStreamProvider initiatorProvider, SwarmUpdateRequest updateRequest, CancellationToken cancellationToken) { if (!SwarmMode) + { + // we still need an active update operation for the TargetVersion + updateOperation = new SwarmUpdateOperation(updateRequest.UpdateVersion); return SwarmPrepareResult.SuccessProviderNotRequired; + } var version = updateRequest.UpdateVersion; var initiator = initiatorProvider != null; logger.LogTrace("PrepareUpdateImpl {version}...", version); - lock (updateSynchronizationLock) - if (version == targetUpdateVersion) - { - logger.LogDebug("Prepare update short circuit!"); - return SwarmPrepareResult.SuccessProviderNotRequired; - } - var shouldAbort = false; + SwarmUpdateOperation localUpdateOperation; try { - lock (updateSynchronizationLock) + SwarmServerResponse sourceNode = null; + lock (swarmServers) { - if (targetUpdateVersion == version) - { - logger.LogTrace("PrepareUpdateFromController early out, already prepared!"); - return SwarmPrepareResult.SuccessProviderNotRequired; - } + var currentNodes = swarmServers + .Select(node => + { + if (node.Identifier == updateRequest.SourceNode) + sourceNode = node; - if (targetUpdateVersion != null) - { - logger.LogWarning("Aborting update preparation, version {targetUpdateVersion} already prepared!", targetUpdateVersion); - shouldAbort = true; - return SwarmPrepareResult.Failure; - } + return node; + }) + .ToList(); + if (swarmController) + localUpdateOperation = new SwarmUpdateOperation( + version, + currentNodes); + else + localUpdateOperation = new SwarmUpdateOperation(version); + } - targetUpdateVersion = version; + var existingUpdateOperation = Interlocked.CompareExchange(ref updateOperation, localUpdateOperation, null); + if (existingUpdateOperation != null && existingUpdateOperation.TargetVersion != version) + { + logger.LogWarning("Aborting update preparation, version {targetUpdateVersion} already prepared!", existingUpdateOperation.TargetVersion); + shouldAbort = true; + return SwarmPrepareResult.Failure; + } + + if (existingUpdateOperation?.TargetVersion == version) + { + logger.LogTrace("PrepareUpdateImpl early out, already prepared!"); + return SwarmPrepareResult.SuccessProviderNotRequired; } if (!swarmController && initiator) @@ -877,11 +876,9 @@ namespace Tgstation.Server.Host.Swarm // File transfer service will hold the necessary streams using var response = await httpClient.SendAsync(request, HttpCompletionOption.ResponseContentRead, cancellationToken); if (response.IsSuccessStatusCode) - { - updateCommitTcs = new TaskCompletionSource(); return SwarmPrepareResult.SuccessHoldProviderUntilCommit; - } + shouldAbort = true; return SwarmPrepareResult.Failure; } @@ -889,29 +886,33 @@ namespace Tgstation.Server.Host.Swarm { logger.LogTrace("Beginning local update process..."); - if (!updateRequest.DownloadTickets.TryGetValue(swarmConfiguration.Identifier, out var ticket)) - { - logger.LogWarning("Missing node entry for download ticket in update request!"); - return SwarmPrepareResult.Failure; - } - - SwarmServerResponse sourceNode; - lock (swarmServers) - sourceNode = swarmServers - .Where(node => node.Identifier == updateRequest.SourceNode) - .FirstOrDefault(); - if (sourceNode == null) { - logger.LogWarning("Missing local node entry for update source node: {sourceNode}", updateRequest.SourceNode); + logger.Log( + swarmController + ? LogLevel.Error + : LogLevel.Warning, + "Missing local node entry for update source node: {sourceNode}", + updateRequest.SourceNode); + shouldAbort = true; + return SwarmPrepareResult.Failure; + } + + if (!updateRequest.DownloadTickets.TryGetValue(swarmConfiguration.Identifier, out var ticket)) + { + logger.Log( + swarmController + ? LogLevel.Error + : LogLevel.Warning, + "Missing node entry for download ticket in update request!"); + shouldAbort = true; return SwarmPrepareResult.Failure; } - var downloaderStream = CreateUpdateStreamProvider(sourceNode, ticket); ServerUpdateResult updateApplyResult; + var downloaderStream = CreateUpdateStreamProvider(sourceNode, ticket); try { - updateCommitTcs = new TaskCompletionSource(); updateApplyResult = await serverUpdater.BeginUpdate( this, downloaderStream, @@ -932,10 +933,7 @@ namespace Tgstation.Server.Host.Swarm } } else - { logger.LogTrace("No need to re-initiate update as it originated here on the swarm controller"); - updateCommitTcs = new TaskCompletionSource(); - } logger.LogDebug("Local node prepared for update to version {version}", version); } @@ -948,56 +946,75 @@ namespace Tgstation.Server.Host.Swarm finally { if (shouldAbort) - await AbortUpdate(cancellationToken); + await AbortUpdate(); } if (!swarmController) return SwarmPrepareResult.SuccessProviderNotRequired; + return await ControllerDistributedPrepareUpdate( + initiatorProvider, + updateRequest, + localUpdateOperation, + cancellationToken); + } + + /// + /// Send a given out to nodes from the swarm controller. + /// + /// The containing the update package if this is the initiating server, otherwise. + /// The . Must always have populated. If is , it must be fully populated. + /// The current . + /// The for the operation. + /// A resulting in the . + async Task ControllerDistributedPrepareUpdate( + ISeekableFileStreamProvider initiatorProvider, + SwarmUpdateRequest updateRequest, + SwarmUpdateOperation currentUpdateOperation, + CancellationToken cancellationToken) + { bool abortUpdate = false; try { logger.LogInformation("Sending remote prepare to nodes..."); - List serversToPrepare; - lock (swarmServers) + + if (currentUpdateOperation.InvolvedServers.Count - 1 < swarmConfiguration.UpdateRequiredNodeCount) { - nodesThatNeedToBeReadyToCommit = new List( - swarmServers - .Where(x => !x.Controller) - .Select(x => x.Identifier)); - - if (nodesThatNeedToBeReadyToCommit.Count < swarmConfiguration.UpdateRequiredNodeCount) - { - logger.LogWarning( - "Aborting update, controller expects to be in sync with {requiredNodeCount} nodes but currently only has {currentNodeCount}!", - swarmConfiguration.UpdateRequiredNodeCount, - nodesThatNeedToBeReadyToCommit.Count); - abortUpdate = true; - return SwarmPrepareResult.Failure; - } - - if (nodesThatNeedToBeReadyToCommit.Count == 0) - { - logger.LogDebug("Controller has no nodes, setting commit-ready."); - var commitTcs = updateCommitTcs; - commitTcs?.TrySetResult(true); - if (commitTcs != null) - return SwarmPrepareResult.SuccessProviderNotRequired; - - logger.LogDebug("Update appears to have been aborted"); - return SwarmPrepareResult.Failure; - } - - serversToPrepare = swarmServers - .Where(x => !x.Controller) - .ToList(); + logger.LogWarning( + "Aborting update, controller expects to be in sync with {requiredNodeCount} nodes but currently only has {currentNodeCount}!", + swarmConfiguration.UpdateRequiredNodeCount, + currentUpdateOperation.InvolvedServers.Count - 1); + abortUpdate = true; + return SwarmPrepareResult.Failure; } - var downloadTicketDictionary = initiator + var weAreInitiator = initiatorProvider != null; + if (currentUpdateOperation.InvolvedServers.Count == 1) + { + logger.LogDebug("Controller has no nodes, setting commit-ready."); + if (updateOperation?.Commit() == true) + return SwarmPrepareResult.SuccessProviderNotRequired; + + logger.LogDebug("Update appears to have been aborted"); + return SwarmPrepareResult.Failure; + } + + // The initiator node obviously doesn't create a ticket for itself + else if (!weAreInitiator && updateRequest.DownloadTickets.Count != currentUpdateOperation.InvolvedServers.Count - 1) + { + logger.LogWarning( + "Aborting update, {receivedTickets} download tickets were provided but there are {nodesToUpdate} nodes in the swarm that require the package!", + updateRequest.DownloadTickets.Count, + currentUpdateOperation.InvolvedServers.Count); + abortUpdate = true; + return SwarmPrepareResult.Failure; + } + + var downloadTicketDictionary = weAreInitiator ? CreateDownloadTickets(initiatorProvider) : updateRequest.DownloadTickets; - var sourceNode = initiator + var sourceNode = weAreInitiator ? swarmConfiguration.Identifier : updateRequest.SourceNode; @@ -1005,7 +1022,9 @@ namespace Tgstation.Server.Host.Swarm using var transferSemaphore = new SemaphoreSlim(1); bool anyFailed = false; - var updateRequests = serversToPrepare + var updateRequests = currentUpdateOperation + .InvolvedServers + .Where(node => !node.Controller) .Select(node => { // only send the necessary ticket to each node from the controller @@ -1013,7 +1032,7 @@ namespace Tgstation.Server.Host.Swarm if (!downloadTicketDictionary.TryGetValue(node.Identifier, out var ticket) && node.Identifier != sourceNode) { - logger.LogWarning("Missing download ticket for node {missingNodeId}!", node.Identifier); + logger.LogError("Missing download ticket for node {missingNodeId}!", node.Identifier); anyFailed = true; return null; } @@ -1025,7 +1044,7 @@ namespace Tgstation.Server.Host.Swarm var request = new SwarmUpdateRequest { - UpdateVersion = version, + UpdateVersion = updateRequest.UpdateVersion, SourceNode = sourceNode, DownloadTickets = localTicketDictionary, }; @@ -1043,7 +1062,6 @@ namespace Tgstation.Server.Host.Swarm var node = tuple.Item1; var body = tuple.Item2; - // no need to provide every server with every ticket using var request = PrepareSwarmRequest( node, HttpMethod.Put, @@ -1060,8 +1078,8 @@ namespace Tgstation.Server.Host.Swarm // if all succeeds... if (tasks.All(x => x.Result)) { - logger.LogInformation("Distributed prepare for update to version {version} complete.", version); - return initiator + logger.LogInformation("Distributed prepare for update to version {version} complete.", updateRequest.UpdateVersion); + return weAreInitiator ? SwarmPrepareResult.SuccessHoldProviderUntilCommit : SwarmPrepareResult.SuccessProviderNotRequired; } @@ -1076,7 +1094,7 @@ namespace Tgstation.Server.Host.Swarm finally { if (abortUpdate) - await AbortUpdate(cancellationToken); + await AbortUpdate(); } return SwarmPrepareResult.Failure; @@ -1218,7 +1236,7 @@ namespace Tgstation.Server.Host.Swarm { logger.LogWarning(ex, "Error during swarm controller health check! Attempting to re-register..."); controllerRegistration = null; - await AbortUpdate(cancellationToken); + await AbortUpdate(); } SwarmRegistrationResult registrationResult; diff --git a/src/Tgstation.Server.Host/Swarm/SwarmUpdateAbortResult.cs b/src/Tgstation.Server.Host/Swarm/SwarmUpdateAbortResult.cs new file mode 100644 index 0000000000..06b4d083a3 --- /dev/null +++ b/src/Tgstation.Server.Host/Swarm/SwarmUpdateAbortResult.cs @@ -0,0 +1,23 @@ +namespace Tgstation.Server.Host.Swarm +{ + /// + /// Result of attempting to abort a . + /// + public enum SwarmUpdateAbortResult + { + /// + /// The operation was successfully aborted by the caller and followup actions should be perform. + /// + Aborted, + + /// + /// The operation was already successfully aborted by another caller. + /// + AlreadyAborted, + + /// + /// The operation cannot abort because it has committed. + /// + CantAbortCommitted, + } +} diff --git a/src/Tgstation.Server.Host/Swarm/SwarmUpdateOperation.cs b/src/Tgstation.Server.Host/Swarm/SwarmUpdateOperation.cs new file mode 100644 index 0000000000..4e7232961b --- /dev/null +++ b/src/Tgstation.Server.Host/Swarm/SwarmUpdateOperation.cs @@ -0,0 +1,128 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; + +using Tgstation.Server.Api.Models.Internal; +using Tgstation.Server.Api.Models.Response; + +namespace Tgstation.Server.Host.Swarm +{ + /// + /// Represents the state of a distributed swarm update. + /// + public class SwarmUpdateOperation + { + /// + /// All of the s that are involved in the updates. + /// + public IReadOnlyList InvolvedServers => initialInvolvedServers ?? throw new InvalidOperationException("This property can only be checked on controller SwarmUpdateOperations!"); + + /// + /// The being updated to. + /// + public Version TargetVersion { get; } + + /// + /// The that represents the final commit. If it results in the update has been committed to and can no longer be aborted. If it results in , it has been aborted. + /// + public Task CommitGate => commitTcs.Task; + + /// + /// Backing field for . + /// + readonly IReadOnlyList initialInvolvedServers; + + /// + /// The backing for . + /// + readonly TaskCompletionSource commitTcs; + + /// + /// of that need to send a ready-commit to the controller before the commit can happen. + /// + readonly HashSet nodesThatNeedToBeReadyToCommit; + + /// + /// Initializes a new instance of the class. + /// + /// The value of . + /// This is the variant for use by non-controller nodes. + public SwarmUpdateOperation(Version targetVersion) + { + TargetVersion = targetVersion ?? throw new ArgumentNullException(nameof(targetVersion)); + commitTcs = new TaskCompletionSource(); + } + + /// + /// Initializes a new instance of the class. + /// + /// The value of . + /// An of the controller's current nodes as s. + /// This is the variant for use by the controller. + public SwarmUpdateOperation(Version targetVersion, IEnumerable currentNodes) + : this(targetVersion) + { + initialInvolvedServers = currentNodes?.ToList() ?? throw new ArgumentNullException(nameof(currentNodes)); + nodesThatNeedToBeReadyToCommit = initialInvolvedServers + .Where(node => !node.Controller) + .Select(node => node.Identifier) + .ToHashSet(); + } + + /// + /// Attempt to abort the update operation. + /// + /// The . + public SwarmUpdateAbortResult Abort() + { + if (commitTcs.TrySetResult(false)) + return SwarmUpdateAbortResult.Aborted; + + return commitTcs.Task.Result + ? SwarmUpdateAbortResult.CantAbortCommitted + : SwarmUpdateAbortResult.AlreadyAborted; + } + + /// + /// Attempt to commit the update. + /// + /// if the commit was successful, if the commit already happened. + public bool Commit() + { + if (nodesThatNeedToBeReadyToCommit != null && nodesThatNeedToBeReadyToCommit.Count != 0) + throw new InvalidOperationException($"Cannot commit! There are still {nodesThatNeedToBeReadyToCommit.Count} nodes that need to be ready!"); + + if (commitTcs.Task.IsCompleted && !commitTcs.Task.Result) + return false; + + commitTcs.SetResult(true); // let the InvalidOperationException throw + return true; + } + + /// + /// Marks a identified by as ready to commit. + /// + /// The to mark as ready. + /// on success, if the update is aborting. + public bool MarkNodeReady(string nodeIdentifier) + { + if (nodeIdentifier == null) + throw new ArgumentNullException(nameof(nodeIdentifier)); + + if (nodesThatNeedToBeReadyToCommit == null) + throw new InvalidOperationException("A non-controller node tried to mark a node as ready!"); + + lock (nodesThatNeedToBeReadyToCommit) + { + if (!nodesThatNeedToBeReadyToCommit.Remove(nodeIdentifier)) + return Abort() != SwarmUpdateAbortResult.Aborted; + + if (nodesThatNeedToBeReadyToCommit.Count == 0) + commitTcs.TrySetResult(true); + + return true; + } + } + } +} diff --git a/tests/Tgstation.Server.Host.Tests/Swarm/SwarmRpcMapper.cs b/tests/Tgstation.Server.Host.Tests/Swarm/SwarmRpcMapper.cs index 98f2d94067..327c973739 100644 --- a/tests/Tgstation.Server.Host.Tests/Swarm/SwarmRpcMapper.cs +++ b/tests/Tgstation.Server.Host.Tests/Swarm/SwarmRpcMapper.cs @@ -186,17 +186,20 @@ namespace Tgstation.Server.Host.Swarm.Tests if (controllerMethod.ReturnType != typeof(IActionResult)) { Assert.AreEqual(typeof(Task), controllerMethod.ReturnType); - args.Add(cancellationToken); + var lastParam = controllerMethod.GetParameters().LastOrDefault(); + if (lastParam?.ParameterType == typeof(CancellationToken)) + args.Add(cancellationToken); + var invocationTask = (Task)controllerMethod.Invoke(controller, args.ToArray()); result = await invocationTask; } else { result = (IActionResult)controllerMethod.Invoke(controller, args.ToArray()); - - // simulate worst case, request completed but was aborted before server replied - cancellationToken.ThrowIfCancellationRequested(); } + + // simulate worst case, request completed but was aborted before server replied + cancellationToken.ThrowIfCancellationRequested(); } else { diff --git a/tests/Tgstation.Server.Host.Tests/Swarm/TestableSwarmNode.cs b/tests/Tgstation.Server.Host.Tests/Swarm/TestableSwarmNode.cs index c296aead63..325abde4d2 100644 --- a/tests/Tgstation.Server.Host.Tests/Swarm/TestableSwarmNode.cs +++ b/tests/Tgstation.Server.Host.Tests/Swarm/TestableSwarmNode.cs @@ -70,10 +70,7 @@ namespace Tgstation.Server.Host.Swarm.Tests foreach (var node in nodes) { - if (node.Config.ControllerAddress != null) - node.Config.UpdateRequiredNodeCount = 0; - else - node.Config.UpdateRequiredNodeCount = (uint)nodes.Length - 1; + node.Config.UpdateRequiredNodeCount = (uint)nodes.Length - 1; node.RpcMapper.Register(configControllerSet); } }