Add SignalR

- Add one hub `/hubs/jobs`, strongly typed API included.
- Adjust request pipeline to support SignalR.
- Allow `Accept: text/event-stream` for SSE requests.
- Add client library support for hubs.
- Add integration tests.
- Add IPermissionSetNotifyee to support dynamic changes based on perms.
- Keep job state in `JobService` for pushing updates.

src/Tgstation.Server.Api/ApiHeaders.cs

@@ -58,6 +58,11 @@ namespace Tgstation.Server.Api
 		/// </summary>
 		public const string ApplicationJsonMime = "application/json";
 
+		/// <summary>
+		/// Added to <see cref="MediaTypeNames.Application"/> in netstandard2.1. Can't use because of lack of .NET Framework support.
+		/// </summary>
+		const string TextEventStreamMime = "text/event-stream";
+
 		/// <summary>
 		/// Get the version of the <see cref="Api"/> the caller is using.
 		/// </summary>
@@ -184,9 +189,10 @@ namespace Tgstation.Server.Api
 		/// </summary>
 		/// <param name="requestHeaders">The <see cref="RequestHeaders"/> containing the serialized <see cref="ApiHeaders"/>.</param>
 		/// <param name="ignoreMissingAuth">If a missing <see cref="HeaderNames.Authorization"/> should be ignored.</param>
+		/// <param name="allowEventStreamAccept">If <see cref="TextEventStreamMime"/> is a valid accept.</param>
 		/// <exception cref="HeadersException">Thrown if the <paramref name="requestHeaders"/> constitue invalid <see cref="ApiHeaders"/>.</exception>
 #pragma warning disable CA1502 // TODO: Decomplexify
-		public ApiHeaders(RequestHeaders requestHeaders, bool ignoreMissingAuth)
+		public ApiHeaders(RequestHeaders requestHeaders, bool ignoreMissingAuth, bool allowEventStreamAccept)
 		{
 			if (requestHeaders == null)
 				throw new ArgumentNullException(nameof(requestHeaders));
@@ -207,8 +213,12 @@ namespace Tgstation.Server.Api
 			}
 
 			var jsonAccept = new Microsoft.Net.Http.Headers.MediaTypeHeaderValue(ApplicationJsonMime);
-			if (!requestHeaders.Accept.Any(x => jsonAccept.IsSubsetOf(x)))
-				AddError(HeaderErrorTypes.Accept, $"Client does not accept {ApplicationJsonMime}!");
+			var eventStreamAccept = new Microsoft.Net.Http.Headers.MediaTypeHeaderValue(TextEventStreamMime);
+			if (!requestHeaders.Accept.Any(jsonAccept.IsSubsetOf))
+				if (!allowEventStreamAccept)
+					AddError(HeaderErrorTypes.Accept, $"Client does not accept {ApplicationJsonMime}!");
+				else if (!requestHeaders.Accept.Any(eventStreamAccept.IsSubsetOf))
+					AddError(HeaderErrorTypes.Accept, $"Client does not accept {ApplicationJsonMime} or {TextEventStreamMime}!");
 
 			if (!requestHeaders.Headers.TryGetValue(HeaderNames.UserAgent, out var userAgentValues) || userAgentValues.Count == 0)
 				AddError(HeaderErrorTypes.UserAgent, $"Missing {HeaderNames.UserAgent} header!");
@@ -386,6 +396,20 @@ namespace Tgstation.Server.Api
 				headers.Add(InstanceIdHeader, instanceId.Value.ToString(CultureInfo.InvariantCulture));
 		}
 
+		/// <summary>
+		/// Adds the <paramref name="headers"/> necessary for a SignalR hub connection.
+		/// </summary>
+		/// <param name="headers">The headers <see cref="IDictionary{TKey, TValue}"/> to write to.</param>
+		public void SetHubConnectionHeaders(IDictionary<string, string> headers)
+		{
+			if (headers == null)
+				throw new ArgumentNullException(nameof(headers));
+
+			headers.Add(HeaderNames.UserAgent, RawUserAgent ?? throw new InvalidOperationException("Missing UserAgent!"));
+			headers.Add(HeaderNames.Accept, ApplicationJsonMime);
+			headers.Add(ApiVersionHeader, CreateApiVersionHeader());
+		}
+
 		/// <summary>
 		/// Create the <see cref="string"/>ified for of the <see cref="ApiVersionHeader"/>.
 		/// </summary>

src/Tgstation.Server.Api/Hubs/ConnectionAbortReason.cs

@@ -0,0 +1,18 @@
+namespace Tgstation.Server.Api.Hubs
+{
+	/// <summary>
+	/// The reason an <see cref="IErrorHandlingHub"/> aborts a connection.
+	/// </summary>
+	public enum ConnectionAbortReason
+	{
+		/// <summary>
+		/// The provided token is no longer authenticated or authorized to keep the connection.
+		/// </summary>
+		TokenInvalid,
+
+		/// <summary>
+		/// The server is restarting.
+		/// </summary>
+		ServerRestart,
+	}
+}

src/Tgstation.Server.Api/Hubs/IErrorHandlingHub.cs

@@ -0,0 +1,19 @@
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Tgstation.Server.Api.Hubs
+{
+	/// <summary>
+	/// Hub for handling communication errors.
+	/// </summary>
+	public interface IErrorHandlingHub
+	{
+		/// <summary>
+		/// Called if a hub connection or call is attempted with an invalid or unauthorized token. After calling this, the connection is aborted.
+		/// </summary>
+		/// <param name="reason">The <see cref="ConnectionAbortReason"/>.</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>A <see cref="ValueTask"/> representing the running operation.</returns>
+		Task AbortingConnection(ConnectionAbortReason reason, CancellationToken cancellationToken);
+	}
+}

src/Tgstation.Server.Api/Hubs/IJobsHub.cs

@@ -0,0 +1,21 @@
+using System.Threading;
+using System.Threading.Tasks;
+
+using Tgstation.Server.Api.Models.Response;
+
+namespace Tgstation.Server.Api.Hubs
+{
+	/// <summary>
+	/// SignalR client methods for receiving <see cref="JobResponse"/>s.
+	/// </summary>
+	public interface IJobsHub : IErrorHandlingHub
+	{
+		/// <summary>
+		/// Push a <paramref name="job"/> update to the client.
+		/// </summary>
+		/// <param name="job">The <see cref="JobResponse"/> to push.</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>A <see cref="ValueTask"/> representing the running operation.</returns>
+		Task ReceiveJobUpdate(JobResponse job, CancellationToken cancellationToken);
+	}
+}

src/Tgstation.Server.Api/Routes.cs

@@ -12,6 +12,11 @@ namespace Tgstation.Server.Api
 		/// </summary>
 		public const string Root = "/";
 
+		/// <summary>
+		/// The root route of all hubs.
+		/// </summary>
+		public const string HubsRoot = Root + "hubs";
+
 		/// <summary>
 		/// The server administration controller.
 		/// </summary>
@@ -102,6 +107,11 @@ namespace Tgstation.Server.Api
 		/// </summary>
 		public const string List = "List";
 
+		/// <summary>
+		/// The root route of all hubs.
+		/// </summary>
+		public const string JobsHub = HubsRoot + "/jobs";
+
 		/// <summary>
 		/// Apply an <paramref name="id"/> postfix to a <paramref name="route"/>.
 		/// </summary>

src/Tgstation.Server.Client/ApiClient.cs

@@ -11,6 +11,10 @@ using System.Threading;
 using System.Threading.Tasks;
 using System.Web;
 
+using Microsoft.AspNetCore.Http.Connections;
+using Microsoft.AspNetCore.SignalR.Client;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
 using Microsoft.Net.Http.Headers;
 
 using Newtonsoft.Json;
@@ -20,6 +24,7 @@ using Newtonsoft.Json.Serialization;
 using Tgstation.Server.Api;
 using Tgstation.Server.Api.Models;
 using Tgstation.Server.Api.Models.Response;
+using Tgstation.Server.Client.Extensions;
 using Tgstation.Server.Common.Extensions;
 using Tgstation.Server.Common.Http;
 
@@ -51,6 +56,18 @@ namespace Tgstation.Server.Client
 			set => httpClient.Timeout = value;
 		}
 
+		/// <summary>
+		/// The <see cref="JsonSerializerSettings"/> to use.
+		/// </summary>
+		static readonly JsonSerializerSettings SerializerSettings = new ()
+		{
+			ContractResolver = new CamelCasePropertyNamesContractResolver(),
+			Converters = new[]
+			{
+				new VersionConverter(),
+			},
+		};
+
 		/// <summary>
 		/// The <see cref="IHttpClient"/> for the <see cref="ApiClient"/>.
 		/// </summary>
@@ -61,6 +78,11 @@ namespace Tgstation.Server.Client
 		/// </summary>
 		readonly List<IRequestLogger> requestLoggers;
 
+		/// <summary>
+		/// List of <see cref="HubConnection"/>s created by the <see cref="ApiClient"/>.
+		/// </summary>
+		readonly List<HubConnection> hubConnections;
+
 		/// <summary>
 		/// Backing field for <see cref="Headers"/>.
 		/// </summary>
@@ -82,14 +104,9 @@ namespace Tgstation.Server.Client
 		ApiHeaders headers;
 
 		/// <summary>
-		/// Get the <see cref="JsonSerializerSettings"/> to use.
+		/// If the <see cref="ApiClient"/> is disposed.
 		/// </summary>
-		/// <returns>A new <see cref="JsonSerializerSettings"/> instance.</returns>
-		static JsonSerializerSettings GetSerializerSettings() => new ()
-		{
-			ContractResolver = new CamelCasePropertyNamesContractResolver(),
-			Converters = new[] { new VersionConverter() },
-		};
+		bool disposed;
 
 		/// <summary>
 		/// Handle a bad HTTP <paramref name="response"/>.
@@ -102,7 +119,7 @@ namespace Tgstation.Server.Client
 			try
 			{
 				// check if json serializes to an error message
-				errorMessage = JsonConvert.DeserializeObject<ErrorMessageResponse>(json, GetSerializerSettings());
+				errorMessage = JsonConvert.DeserializeObject<ErrorMessageResponse>(json, SerializerSettings);
 			}
 			catch (JsonException)
 			{
@@ -149,7 +166,12 @@ namespace Tgstation.Server.Client
 		/// <param name="apiHeaders">The value of <see cref="Headers"/>.</param>
 		/// <param name="tokenRefreshHeaders">The value of <see cref="tokenRefreshHeaders"/>.</param>
 		/// <param name="authless">The value of <see cref="authless"/>.</param>
-		public ApiClient(IHttpClient httpClient, Uri url, ApiHeaders apiHeaders, ApiHeaders? tokenRefreshHeaders, bool authless)
+		public ApiClient(
+			IHttpClient httpClient,
+			Uri url,
+			ApiHeaders apiHeaders,
+			ApiHeaders? tokenRefreshHeaders,
+			bool authless)
 		{
 			this.httpClient = httpClient ?? throw new ArgumentNullException(nameof(httpClient));
 			Url = url ?? throw new ArgumentNullException(nameof(url));
@@ -158,12 +180,27 @@ namespace Tgstation.Server.Client
 			this.authless = authless;
 
 			requestLoggers = new List<IRequestLogger>();
+			hubConnections = new List<HubConnection>();
 			semaphoreSlim = new SemaphoreSlim(1);
 		}
 
 		/// <inheritdoc />
-		public void Dispose()
+		public async ValueTask DisposeAsync()
 		{
+			List<HubConnection> localHubConnections;
+			lock (hubConnections)
+			{
+				if (disposed)
+					return;
+
+				disposed = true;
+
+				localHubConnections = hubConnections.ToList();
+				hubConnections.Clear();
+			}
+
+			await ValueTaskExtensions.WhenAll(hubConnections.Select(connection => connection.DisposeAsync()));
+
 			httpClient.Dispose();
 			semaphoreSlim.Dispose();
 		}
@@ -294,6 +331,128 @@ namespace Tgstation.Server.Client
 			}
 		}
 
+		/// <summary>
+		/// Attempt to refresh the stored Bearer token in <see cref="Headers"/>.
+		/// </summary>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>A <see cref="ValueTask{TResult}"/> resulting in <see langword="true"/> if the refresh was successful, <see langword="false"/> if a refresh is unable to be performed.</returns>
+		public async ValueTask<bool> RefreshToken(CancellationToken cancellationToken)
+		{
+			if (tokenRefreshHeaders == null)
+				return false;
+
+			var startingToken = headers.Token;
+			await semaphoreSlim.WaitAsync(cancellationToken).ConfigureAwait(false);
+			try
+			{
+				if (startingToken != headers.Token)
+					return true;
+
+				var token = await RunRequest<object, TokenResponse>(Routes.Root, new object(), HttpMethod.Post, null, true, cancellationToken).ConfigureAwait(false);
+				headers = new ApiHeaders(headers.UserAgent!, token);
+			}
+			finally
+			{
+				semaphoreSlim.Release();
+			}
+
+			return true;
+		}
+
+		/// <inheritdoc />
+		public async ValueTask<IAsyncDisposable> CreateHubConnection<THubImplementation>(
+			THubImplementation hubImplementation,
+			IRetryPolicy? retryPolicy,
+			Action<ILoggingBuilder>? loggingConfigureAction,
+			CancellationToken cancellationToken)
+			where THubImplementation : class
+		{
+			if (hubImplementation == null)
+				throw new ArgumentNullException(nameof(hubImplementation));
+
+			retryPolicy ??= new InfiniteThirtySecondMaxRetryPolicy();
+
+			HubConnection? hubConnection = null;
+			var hubConnectionBuilder = new HubConnectionBuilder()
+				.AddNewtonsoftJsonProtocol(options =>
+				{
+					options.PayloadSerializerSettings = SerializerSettings;
+				})
+				.WithAutomaticReconnect(retryPolicy)
+				.WithUrl(
+					new Uri(Url, Routes.JobsHub),
+					HttpTransportType.ServerSentEvents,
+					options =>
+					{
+						options.AccessTokenProvider = async () =>
+						{
+							// DCT: None available.
+							if (Headers.Token == null
+								|| (Headers.Token.ParseJwt().ValidTo <= DateTime.UtcNow
+								&& !await RefreshToken(CancellationToken.None)))
+							{
+								_ = hubConnection!.StopAsync(); // DCT: None available.
+								return null;
+							}
+
+							return Headers.Token.Bearer;
+						};
+
+						options.CloseTimeout = Timeout;
+
+						Headers.SetHubConnectionHeaders(options.Headers);
+					});
+
+			if (loggingConfigureAction != null)
+				hubConnectionBuilder.ConfigureLogging(loggingConfigureAction);
+
+			hubConnection = hubConnectionBuilder.Build();
+			try
+			{
+				hubConnection.Closed += async (error) =>
+				{
+					if (error is HttpRequestException httpRequestException)
+					{
+						// .StatusCode isn't in netstandard but fuck the police
+						var property = error.GetType().GetProperty("StatusCode");
+						if (property != null)
+						{
+							var statusCode = (HttpStatusCode?)property.GetValue(error);
+							if (statusCode == HttpStatusCode.Unauthorized
+								&& !await RefreshToken(CancellationToken.None))
+								_ = hubConnection!.StopAsync();
+						}
+					}
+				};
+
+				hubConnection.ProxyOn(hubImplementation);
+
+				Task startTask;
+				lock (hubConnections)
+				{
+					if (disposed)
+						throw new ObjectDisposedException(nameof(ApiClient));
+
+					hubConnections.Add(hubConnection);
+					startTask = hubConnection.StartAsync(cancellationToken);
+				}
+
+				await startTask;
+
+				return hubConnection;
+			}
+			catch
+			{
+				bool needsDispose;
+				lock (hubConnections)
+					needsDispose = hubConnections.Remove(hubConnection);
+
+				if (needsDispose)
+					await hubConnection.DisposeAsync();
+				throw;
+			}
+		}
+
 		/// <summary>
 		/// Main request method.
 		/// </summary>
@@ -305,6 +464,7 @@ namespace Tgstation.Server.Client
 		/// <param name="tokenRefresh">If this is a token refresh operation.</param>
 		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
 		/// <returns>A <see cref="ValueTask{TResult}"/> resulting in the response on success.</returns>
+#pragma warning disable CA1506 // TODO: Decomplexify
 		protected virtual async ValueTask<TResult> RunRequest<TResult>(
 			string route,
 			HttpContent? content,
@@ -322,7 +482,7 @@ namespace Tgstation.Server.Client
 
 			HttpResponseMessage response;
 			var fullUri = new Uri(Url, route);
-			var serializerSettings = GetSerializerSettings();
+			var serializerSettings = SerializerSettings;
 			var fileDownload = typeof(TResult) == typeof(Stream);
 			using (var request = new HttpRequestMessage(method, fullUri))
 			{
@@ -418,38 +578,7 @@ namespace Tgstation.Server.Client
 				}
 			}
 		}
-
-		/// <summary>
-		/// Attempt to refresh the bearer token in the <see cref="headers"/>.
-		/// </summary>
-		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
-		/// <returns>A <see cref="ValueTask{TResult}"/> resulting in <see langword="true"/> if the refresh was successful, <see langword="false"/> otherwise.</returns>
-		async ValueTask<bool> RefreshToken(CancellationToken cancellationToken)
-		{
-			if (tokenRefreshHeaders == null)
-				return false;
-
-			var startingToken = headers.Token;
-			await semaphoreSlim.WaitAsync(cancellationToken).ConfigureAwait(false);
-			try
-			{
-				if (startingToken != headers.Token)
-					return true;
-
-				var token = await RunRequest<object, TokenResponse>(Routes.Root, new object(), HttpMethod.Post, null, true, cancellationToken).ConfigureAwait(false);
-				headers = new ApiHeaders(headers.UserAgent!, token);
-			}
-			catch (ClientException)
-			{
-				return false;
-			}
-			finally
-			{
-				semaphoreSlim.Release();
-			}
-
-			return true;
-		}
+#pragma warning restore CA1506
 
 		/// <summary>
 		/// Main request method.
@@ -475,7 +604,7 @@ namespace Tgstation.Server.Client
 			HttpContent? content = null;
 			if (body != null)
 				content = new StringContent(
-					JsonConvert.SerializeObject(body, typeof(TBody), Formatting.None, GetSerializerSettings()),
+					JsonConvert.SerializeObject(body, typeof(TBody), Formatting.None, SerializerSettings),
 					Encoding.UTF8,
 					ApiHeaders.ApplicationJsonMime);
 

src/Tgstation.Server.Client/Extensions/HubConnectionExtensions.cs

@@ -0,0 +1,104 @@
+using System;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+using Microsoft.AspNetCore.SignalR.Client;
+
+namespace Tgstation.Server.Client.Extensions
+{
+	/// <summary>
+	/// Extension methods for the <see cref="HubConnection"/> <see langword="class"/>.
+	/// </summary>
+	static class HubConnectionExtensions
+	{
+		/// <summary>
+		/// Apply a given <paramref name="proxy"/> to a given <paramref name="hubConnection"/>.
+		/// </summary>
+		/// <typeparam name="TClientProxy">The strongly typed client proxy.</typeparam>
+		/// <param name="hubConnection">The <see cref="HubConnection"/> to proxy on.</param>
+		/// <param name="proxy">The <typeparamref name="TClientProxy"/> to forward operations to.</param>
+		public static void ProxyOn<TClientProxy>(this HubConnection hubConnection, TClientProxy proxy)
+			where TClientProxy : class
+		{
+			if (hubConnection == null)
+				throw new ArgumentNullException(nameof(hubConnection));
+
+			if (proxy == null)
+				throw new ArgumentNullException(nameof(proxy));
+
+			ProxyOn(hubConnection, typeof(TClientProxy), proxy);
+		}
+
+		/// <summary>
+		/// Apply a given <paramref name="proxyObject"/> to a given <paramref name="hubConnection"/>.
+		/// </summary>
+		/// <param name="hubConnection">The <see cref="HubConnection"/> to proxy on.</param>
+		/// <param name="proxyType">The <see cref="Type"/> of <paramref name="proxyObject"/>.</param>
+		/// <param name="proxyObject">The <see cref="object"/> to forward operations to.</param>
+		static void ProxyOn(this HubConnection hubConnection, Type proxyType, object proxyObject)
+		{
+			var clientMethods = proxyType.GetMethods();
+			var cancellationTokenType = typeof(CancellationToken);
+			foreach (var clientMethod in clientMethods)
+			{
+				var parametersList = clientMethod
+					.GetParameters()
+					.Select(parameterInfo => parameterInfo.ParameterType)
+					.ToList();
+
+				var cancellationTokenIndex = parametersList.IndexOf(cancellationTokenType);
+				if (cancellationTokenIndex != -1)
+				{
+					parametersList.RemoveAt(cancellationTokenIndex);
+#if DEBUG
+					if (parametersList.IndexOf(cancellationTokenType) != -1)
+						throw new InvalidOperationException("Cannot ProxyOn a method with multiple CancellationToken parameters!");
+#endif
+				}
+
+				var parameters = parametersList.ToArray();
+
+				object?[] AddCancellationTokenToParametersArray(object?[] parametersArray)
+				{
+					if (cancellationTokenIndex == -1)
+						return parametersArray;
+
+					var newList = parametersArray.ToList();
+					newList.Insert(cancellationTokenIndex, CancellationToken.None);
+					return newList.ToArray();
+				}
+
+				var returnType = clientMethod.ReturnType;
+				if (returnType != typeof(Task))
+				{
+					if (returnType.BaseType != typeof(Task))
+						throw new InvalidOperationException($"Return type {returnType} of {proxyType.FullName}.{clientMethod.Name} is not supported! Only Task and derivatives are supported.");
+
+					var resultProperty = returnType.GetProperty(nameof(Task<object>.Result));
+					hubConnection.On(
+						clientMethod.Name,
+						parameters,
+						async (parameterArray, _) =>
+						{
+							var task = (Task)clientMethod.Invoke(proxyObject, AddCancellationTokenToParametersArray(parameterArray));
+							await task;
+							return resultProperty.GetValue(task);
+						},
+						hubConnection);
+				}
+				else
+					hubConnection.On(
+						clientMethod.Name,
+						parameters,
+						(parameterArray) =>
+						{
+							return (Task)clientMethod.Invoke(proxyObject, AddCancellationTokenToParametersArray(parameterArray));
+						});
+			}
+
+			foreach (var inheritedInterface in proxyType.GetInterfaces())
+				ProxyOn(hubConnection, inheritedInterface, proxyObject);
+		}
+	}
+}

src/Tgstation.Server.Client/IApiClient.cs

@@ -3,6 +3,9 @@ using System.IO;
 using System.Threading;
 using System.Threading.Tasks;
 
+using Microsoft.AspNetCore.SignalR.Client;
+using Microsoft.Extensions.Logging;
+
 using Tgstation.Server.Api;
 using Tgstation.Server.Api.Models;
 using Tgstation.Server.Api.Models.Response;
@@ -12,7 +15,7 @@ namespace Tgstation.Server.Client
 	/// <summary>
 	/// Web interface for the API.
 	/// </summary>
-	interface IApiClient : IDisposable
+	interface IApiClient : IAsyncDisposable
 	{
 		/// <summary>
 		/// The <see cref="ApiHeaders"/> the <see cref="IApiClient"/> uses.
@@ -35,6 +38,22 @@ namespace Tgstation.Server.Client
 		/// <param name="requestLogger">The <see cref="IRequestLogger"/> to add.</param>
 		void AddRequestLogger(IRequestLogger requestLogger);
 
+		/// <summary>
+		/// Subscribe to all job updates available to the <see cref="IServerClient"/>.
+		/// </summary>
+		/// <typeparam name="THubImplementation">The <see cref="Type"/> of the hub being implemented.</typeparam>
+		/// <param name="hubImplementation">The <typeparamref name="THubImplementation"/> to use for proxying the methods of the hub connection.</param>
+		/// <param name="retryPolicy">The optional <see cref="IRetryPolicy"/> to use for the backing connection. The default retry policy waits for 1, 2, 4, 8, and 16 seconds, then 30s repeatedly.</param>
+		/// <param name="loggingConfigureAction">The optional <see cref="Action{T1}"/> used to configure a <see cref="ILoggingBuilder"/>.</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>An <see cref="IAsyncDisposable"/> representing the lifetime of the subscription.</returns>
+		ValueTask<IAsyncDisposable> CreateHubConnection<THubImplementation>(
+			THubImplementation hubImplementation,
+			IRetryPolicy? retryPolicy,
+			Action<ILoggingBuilder>? loggingConfigureAction,
+			CancellationToken cancellationToken)
+			where THubImplementation : class;
+
 		/// <summary>
 		/// Run an HTTP PUT request.
 		/// </summary>

src/Tgstation.Server.Client/IServerClient.cs

@@ -2,6 +2,10 @@
 using System.Threading;
 using System.Threading.Tasks;
 
+using Microsoft.AspNetCore.SignalR.Client;
+using Microsoft.Extensions.Logging;
+
+using Tgstation.Server.Api.Hubs;
 using Tgstation.Server.Api.Models.Response;
 
 namespace Tgstation.Server.Client
@@ -9,7 +13,7 @@ namespace Tgstation.Server.Client
 	/// <summary>
 	/// Main client for communicating with a server.
 	/// </summary>
-	public interface IServerClient : IDisposable
+	public interface IServerClient : IAsyncDisposable
 	{
 		/// <summary>
 		/// The connected server <see cref="Uri"/>.
@@ -53,6 +57,20 @@ namespace Tgstation.Server.Client
 		/// <returns>A <see cref="Task{TResult}"/> resulting in the <see cref="ServerInformationResponse"/> of the target server.</returns>
 		ValueTask<ServerInformationResponse> ServerInformation(CancellationToken cancellationToken);
 
+		/// <summary>
+		/// Subscribe to all job updates available to the <see cref="IServerClient"/>.
+		/// </summary>
+		/// <param name="jobsReceiver">The <see cref="IJobsHub"/> to use to subscribe to updates.</param>
+		/// <param name="retryPolicy">The optional <see cref="IRetryPolicy"/> to use for the backing connection. The default retry policy waits for 1, 2, 4, 8, and 16 seconds, then 30s repeatedly.</param>
+		/// <param name="loggingConfigureAction">The optional <see cref="Action{T1}"/> used to configure a <see cref="ILoggingBuilder"/>.</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>An <see cref="IAsyncDisposable"/> representing the lifetime of the subscription.</returns>
+		ValueTask<IAsyncDisposable> SubscribeToJobUpdates(
+			IJobsHub jobsReceiver,
+			IRetryPolicy? retryPolicy = null,
+			Action<ILoggingBuilder>? loggingConfigureAction = null,
+			CancellationToken cancellationToken = default);
+
 		/// <summary>
 		/// Adds a <paramref name="requestLogger"/> to the request pipeline.
 		/// </summary>

src/Tgstation.Server.Client/InfiniteThirtySecondMaxRetryPolicy.cs

@@ -0,0 +1,21 @@
+using System;
+
+using Microsoft.AspNetCore.SignalR.Client;
+
+namespace Tgstation.Server.Client
+{
+	/// <summary>
+	/// A <see cref="IRetryPolicy"/> that returns seconds in powers of 2, maxing out at 30s.
+	/// </summary>
+	sealed class InfiniteThirtySecondMaxRetryPolicy : IRetryPolicy
+	{
+		/// <inheritdoc />
+		public TimeSpan? NextRetryDelay(RetryContext retryContext)
+		{
+			if (retryContext == null)
+				throw new ArgumentNullException(nameof(retryContext));
+
+			return TimeSpan.FromSeconds(Math.Min(Math.Pow(2, retryContext.PreviousRetryCount), 30));
+		}
+	}
+}

src/Tgstation.Server.Client/ServerClient.cs

@@ -2,7 +2,11 @@
 using System.Threading;
 using System.Threading.Tasks;
 
+using Microsoft.AspNetCore.SignalR.Client;
+using Microsoft.Extensions.Logging;
+
 using Tgstation.Server.Api;
+using Tgstation.Server.Api.Hubs;
 using Tgstation.Server.Api.Models.Response;
 
 namespace Tgstation.Server.Client
@@ -59,12 +63,20 @@ namespace Tgstation.Server.Client
 		}
 
 		/// <inheritdoc />
-		public void Dispose() => apiClient.Dispose();
+		public ValueTask DisposeAsync() => apiClient.DisposeAsync();
 
 		/// <inheritdoc />
 		public ValueTask<ServerInformationResponse> ServerInformation(CancellationToken cancellationToken) => apiClient.Read<ServerInformationResponse>(Routes.Root, cancellationToken);
 
 		/// <inheritdoc />
 		public void AddRequestLogger(IRequestLogger requestLogger) => apiClient.AddRequestLogger(requestLogger);
+
+		/// <inheritdoc />
+		public ValueTask<IAsyncDisposable> SubscribeToJobUpdates(
+			IJobsHub jobsReceiver,
+			IRetryPolicy? retryPolicy,
+			Action<ILoggingBuilder>? loggingConfigureAction,
+			CancellationToken cancellationToken)
+			=> apiClient.CreateHubConnection(jobsReceiver, retryPolicy, loggingConfigureAction, cancellationToken);
 	}
 }

src/Tgstation.Server.Client/ServerClientFactory.cs

@@ -120,7 +120,7 @@ namespace Tgstation.Server.Client
 			TimeSpan? timeout = null,
 			CancellationToken cancellationToken = default)
 		{
-			using var api = ApiClientFactory.CreateApiClient(
+			await using var api = ApiClientFactory.CreateApiClient(
 				host,
 				new ApiHeaders(
 					productHeaderValue,
@@ -162,7 +162,7 @@ namespace Tgstation.Server.Client
 			requestLoggers ??= Enumerable.Empty<IRequestLogger>();
 
 			TokenResponse token;
-			using (var api = ApiClientFactory.CreateApiClient(host, loginHeaders, null, false))
+			await using (var api = ApiClientFactory.CreateApiClient(host, loginHeaders, null, false))
 			{
 				foreach (var requestLogger in requestLoggers)
 					api.AddRequestLogger(requestLogger);

src/Tgstation.Server.Client/Tgstation.Server.Client.csproj

@@ -9,6 +9,13 @@
     <PackageReleaseNotes>$(TGS_NUGET_RELEASE_NOTES_CLIENT)</PackageReleaseNotes>
   </PropertyGroup>
 
+  <ItemGroup>
+    <!-- Usage: Connecting to SignalR hubs in API -->
+    <PackageReference Include="Microsoft.AspNetCore.SignalR.Client" Version="7.0.13" />
+    <!-- Usage: Using target JSON serializer for API -->
+    <PackageReference Include="Microsoft.AspNetCore.SignalR.Protocols.NewtonsoftJson" Version="7.0.13" />
+  </ItemGroup>
+
   <ItemGroup>
     <ProjectReference Include="..\Tgstation.Server.Api\Tgstation.Server.Api.csproj" />
     <ProjectReference Include="..\Tgstation.Server.Common\Tgstation.Server.Common.csproj" /> <!-- Needed for explicit nuget versioning -->

src/Tgstation.Server.Host/Components/InstanceWrapper.cs

@@ -1,5 +1,4 @@
-using System;
-using System.Threading;
+using System.Threading;
 using System.Threading.Tasks;
 
 using Tgstation.Server.Host.Components.Byond;

src/Tgstation.Server.Host/Controllers/ApiController.cs

@@ -116,10 +116,6 @@ namespace Tgstation.Server.Host.Controllers
 				if (requireHeaders)
 					return HeadersIssue(ApiHeadersProvider.HeadersException);
 			}
-			else if (!ApiHeaders.Compatible())
-				return this.StatusCode(
-					HttpStatusCode.UpgradeRequired,
-					new ErrorMessageResponse(ErrorCode.ApiMismatch));
 
 			var errorCase = await ValidateRequest(cancellationToken);
 			if (errorCase != null)

src/Tgstation.Server.Host/Controllers/HomeController.cs

@@ -1,6 +1,5 @@
 using System;
 using System.Linq;
-using System.Net;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -21,7 +20,6 @@ using Tgstation.Server.Host.Components.Interop;
 using Tgstation.Server.Host.Configuration;
 using Tgstation.Server.Host.Core;
 using Tgstation.Server.Host.Database;
-using Tgstation.Server.Host.Extensions;
 using Tgstation.Server.Host.Models;
 using Tgstation.Server.Host.Security;
 using Tgstation.Server.Host.Security.OAuth;
@@ -182,11 +180,7 @@ namespace Tgstation.Server.Host.Controllers
 				try
 				{
 					// we only allow authorization header issues
-					var headers = ApiHeadersProvider.CreateAuthlessHeaders();
-					if (!headers.Compatible())
-						return this.StatusCode(
-							HttpStatusCode.UpgradeRequired,
-							new ErrorMessageResponse(ErrorCode.ApiMismatch));
+					ApiHeadersProvider.CreateAuthlessHeaders();
 				}
 				catch (HeadersException ex)
 				{

src/Tgstation.Server.Host/Controllers/InstanceController.cs

@@ -68,6 +68,11 @@ namespace Tgstation.Server.Host.Controllers
 		/// </summary>
 		readonly IPortAllocator portAllocator;
 
+		/// <summary>
+		/// The <see cref="IPermissionsUpdateNotifyee"/> for the <see cref="InstanceController"/>.
+		/// </summary>
+		readonly IPermissionsUpdateNotifyee permissionsUpdateNotifyee;
+
 		/// <summary>
 		/// The <see cref="GeneralConfiguration"/> for the <see cref="InstanceController"/>.
 		/// </summary>
@@ -88,7 +93,8 @@ namespace Tgstation.Server.Host.Controllers
 		/// <param name="jobManager">The value of <see cref="jobManager"/>.</param>
 		/// <param name="ioManager">The value of <see cref="ioManager"/>.</param>
 		/// <param name="platformIdentifier">The value of <see cref="platformIdentifier"/>.</param>
-		/// <param name="portAllocator">The value of <see cref="IPortAllocator"/>.</param>
+		/// <param name="portAllocator">The value of <see cref="portAllocator"/>.</param>
+		/// <param name="permissionsUpdateNotifyee">The value of <see cref="permissionsUpdateNotifyee"/>.</param>
 		/// <param name="generalConfigurationOptions">The <see cref="IOptions{TOptions}"/> containing the value of <see cref="generalConfiguration"/>.</param>
 		/// <param name="swarmConfigurationOptions">The <see cref="IOptions{TOptions}"/> containing the value of <see cref="swarmConfiguration"/>.</param>
 		/// <param name="apiHeaders">The <see cref="IApiHeadersProvider"/> for the <see cref="ComponentInterfacingController"/>.</param>
@@ -101,6 +107,7 @@ namespace Tgstation.Server.Host.Controllers
 			IIOManager ioManager,
 			IPortAllocator portAllocator,
 			IPlatformIdentifier platformIdentifier,
+			IPermissionsUpdateNotifyee permissionsUpdateNotifyee,
 			IOptions<GeneralConfiguration> generalConfigurationOptions,
 			IOptions<SwarmConfiguration> swarmConfigurationOptions,
 			IApiHeadersProvider apiHeaders)
@@ -116,6 +123,8 @@ namespace Tgstation.Server.Host.Controllers
 			this.ioManager = ioManager ?? throw new ArgumentNullException(nameof(ioManager));
 			this.platformIdentifier = platformIdentifier ?? throw new ArgumentNullException(nameof(platformIdentifier));
 			this.portAllocator = portAllocator ?? throw new ArgumentNullException(nameof(portAllocator));
+			this.permissionsUpdateNotifyee = permissionsUpdateNotifyee ?? throw new ArgumentNullException(nameof(permissionsUpdateNotifyee));
+
 			generalConfiguration = generalConfigurationOptions?.Value ?? throw new ArgumentNullException(nameof(generalConfigurationOptions));
 			swarmConfiguration = swarmConfigurationOptions?.Value ?? throw new ArgumentNullException(nameof(swarmConfigurationOptions));
 		}
@@ -267,6 +276,10 @@ namespace Tgstation.Server.Host.Controllers
 				newInstance.Id,
 				newInstance.Path);
 
+			await permissionsUpdateNotifyee.InstancePermissionSetCreated(
+				newInstance.InstancePermissionSets.First(),
+				cancellationToken);
+
 			var api = newInstance.ToApi();
 			api.Accessible = true; // instances are always accessible by their creator
 			return attached ? Json(api) : Created(api);
@@ -770,6 +783,7 @@ namespace Tgstation.Server.Host.Controllers
 		{
 			permissionSetToModify ??= new InstancePermissionSet()
 			{
+				PermissionSet = AuthenticationContext.PermissionSet,
 				PermissionSetId = AuthenticationContext.PermissionSet.Id.Value,
 			};
 			permissionSetToModify.ByondRights = RightsHelper.AllRights<ByondRights>();

src/Tgstation.Server.Host/Controllers/InstancePermissionSetController.cs

@@ -30,19 +30,26 @@ namespace Tgstation.Server.Host.Controllers
 	[Route(Routes.InstancePermissionSet)]
 	public sealed class InstancePermissionSetController : InstanceRequiredController
 	{
+		/// <summary>
+		/// The <see cref="IPermissionsUpdateNotifyee"/> for the <see cref="InstancePermissionSetController"/>.
+		/// </summary>
+		readonly IPermissionsUpdateNotifyee permissionsUpdateNotifyee;
+
 		/// <summary>
 		/// Initializes a new instance of the <see cref="InstancePermissionSetController"/> class.
 		/// </summary>
 		/// <param name="databaseContext">The <see cref="IDatabaseContext"/> for the <see cref="InstanceRequiredController"/>.</param>
-		/// <param name="authenticationContext">The <see cref="IAuthenticationContextFactory"/> for the <see cref="InstanceRequiredController"/>.</param>
+		/// <param name="authenticationContext">The <see cref="IAuthenticationContext"/> for the <see cref="InstanceRequiredController"/>.</param>
 		/// <param name="logger">The <see cref="ILogger"/> for the <see cref="InstanceRequiredController"/>.</param>
 		/// <param name="instanceManager">The <see cref="IInstanceManager"/> for the <see cref="InstanceRequiredController"/>.</param>
+		/// <param name="permissionsUpdateNotifyee">The value of <see cref="permissionsUpdateNotifyee"/>.</param>
 		/// <param name="apiHeaders">The <see cref="IApiHeadersProvider"/> for the <see cref="InstanceRequiredController"/>.</param>
 		public InstancePermissionSetController(
 			IDatabaseContext databaseContext,
 			IAuthenticationContext authenticationContext,
 			ILogger<InstancePermissionSetController> logger,
 			IInstanceManager instanceManager,
+			IPermissionsUpdateNotifyee permissionsUpdateNotifyee,
 			IApiHeadersProvider apiHeaders)
 			: base(
 				  databaseContext,
@@ -51,6 +58,7 @@ namespace Tgstation.Server.Host.Controllers
 				  instanceManager,
 				  apiHeaders)
 		{
+			this.permissionsUpdateNotifyee = permissionsUpdateNotifyee ?? throw new ArgumentNullException(nameof(permissionsUpdateNotifyee));
 		}
 
 		/// <summary>
@@ -76,6 +84,7 @@ namespace Tgstation.Server.Host.Controllers
 				.Where(x => x.Id == model.PermissionSetId)
 				.Select(x => new Models.PermissionSet
 				{
+					Id = x.Id,
 					UserId = x.UserId,
 				})
 				.FirstOrDefaultAsync(cancellationToken);
@@ -112,6 +121,10 @@ namespace Tgstation.Server.Host.Controllers
 			DatabaseContext.InstancePermissionSets.Add(dbUser);
 
 			await DatabaseContext.Save(cancellationToken);
+
+			// needs to be set for next call
+			dbUser.PermissionSet = existingPermissionSet;
+			await permissionsUpdateNotifyee.InstancePermissionSetCreated(dbUser, cancellationToken);
 			return Created(dbUser.ToApi());
 		}
 #pragma warning restore CA1506

src/Tgstation.Server.Host/Controllers/UserController.cs

@@ -40,6 +40,11 @@ namespace Tgstation.Server.Host.Controllers
 		/// </summary>
 		readonly ICryptographySuite cryptographySuite;
 
+		/// <summary>
+		/// The <see cref="IPermissionsUpdateNotifyee"/> for the <see cref="UserController"/>.
+		/// </summary>
+		readonly IPermissionsUpdateNotifyee permissionsUpdateNotifyee;
+
 		/// <summary>
 		/// The <see cref="GeneralConfiguration"/> for the <see cref="UserController"/>.
 		/// </summary>
@@ -52,6 +57,7 @@ namespace Tgstation.Server.Host.Controllers
 		/// <param name="authenticationContext">The <see cref="IAuthenticationContext"/> for the <see cref="ApiController"/>.</param>
 		/// <param name="systemIdentityFactory">The value of <see cref="systemIdentityFactory"/>.</param>
 		/// <param name="cryptographySuite">The value of <see cref="cryptographySuite"/>.</param>
+		/// <param name="permissionsUpdateNotifyee">The value of <see cref="permissionsUpdateNotifyee"/>.</param>
 		/// <param name="logger">The <see cref="ILogger"/> for the <see cref="ApiController"/>.</param>
 		/// <param name="generalConfigurationOptions">The <see cref="IOptions{TOptions}"/> containing the value of <see cref="generalConfiguration"/>.</param>
 		/// <param name="apiHeaders">The <see cref="IApiHeadersProvider"/> for the <see cref="ApiController"/>.</param>
@@ -60,6 +66,7 @@ namespace Tgstation.Server.Host.Controllers
 			IAuthenticationContext authenticationContext,
 			ISystemIdentityFactory systemIdentityFactory,
 			ICryptographySuite cryptographySuite,
+			IPermissionsUpdateNotifyee permissionsUpdateNotifyee,
 			ILogger<UserController> logger,
 			IOptions<GeneralConfiguration> generalConfigurationOptions,
 			IApiHeadersProvider apiHeaders)
@@ -72,6 +79,7 @@ namespace Tgstation.Server.Host.Controllers
 		{
 			this.systemIdentityFactory = systemIdentityFactory ?? throw new ArgumentNullException(nameof(systemIdentityFactory));
 			this.cryptographySuite = cryptographySuite ?? throw new ArgumentNullException(nameof(cryptographySuite));
+			this.permissionsUpdateNotifyee = permissionsUpdateNotifyee ?? throw new ArgumentNullException(nameof(permissionsUpdateNotifyee));
 			generalConfiguration = generalConfigurationOptions?.Value ?? throw new ArgumentNullException(nameof(generalConfigurationOptions));
 		}
 
@@ -243,13 +251,17 @@ namespace Tgstation.Server.Host.Controllers
 			if (model.Name != null && Models.User.CanonicalizeName(model.Name) != originalUser.CanonicalName)
 				return BadRequest(new ErrorMessageResponse(ErrorCode.UserNameChange));
 
+			bool userWasDisabled;
 			if (model.Enabled.HasValue)
 			{
-				if (originalUser.Enabled.Value && !model.Enabled.Value)
+				userWasDisabled = originalUser.Enabled.Value && !model.Enabled.Value;
+				if (userWasDisabled)
 					originalUser.LastPasswordUpdate = DateTimeOffset.UtcNow;
 
 				originalUser.Enabled = model.Enabled.Value;
 			}
+			else
+				userWasDisabled = false;
 
 			if (model.OAuthConnections != null
 				&& (model.OAuthConnections.Count != originalUser.OAuthConnections.Count
@@ -315,6 +327,9 @@ namespace Tgstation.Server.Host.Controllers
 
 			Logger.LogInformation("Updated user {userName} ({userId})", originalUser.Name, originalUser.Id);
 
+			if (userWasDisabled)
+				await permissionsUpdateNotifyee.UserDisabled(originalUser, cancellationToken);
+
 			// return id only if not a self update and cannot read users
 			var canReadBack = AuthenticationContext.User.Id == originalUser.Id
 				|| callerAdministrationRights.HasFlag(AdministrationRights.ReadUsers);

src/Tgstation.Server.Host/Core/Application.cs

@@ -14,9 +14,12 @@ using Microsoft.AspNetCore.Authentication.JwtBearer;
 using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Cors.Infrastructure;
 using Microsoft.AspNetCore.Hosting;
+using Microsoft.AspNetCore.Http;
+using Microsoft.AspNetCore.Http.Connections;
 using Microsoft.AspNetCore.Identity;
 using Microsoft.AspNetCore.Mvc.Filters;
 using Microsoft.AspNetCore.Mvc.Infrastructure;
+using Microsoft.AspNetCore.SignalR;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
@@ -31,6 +34,7 @@ using Serilog.Formatting.Display;
 using Serilog.Sinks.Elasticsearch;
 
 using Tgstation.Server.Api;
+using Tgstation.Server.Api.Hubs;
 using Tgstation.Server.Common.Http;
 using Tgstation.Server.Host.Components;
 using Tgstation.Server.Host.Components.Byond;
@@ -250,6 +254,18 @@ namespace Tgstation.Server.Host.Core
 					ConfigureNewtonsoftJsonSerializerSettingsForApi(options.SerializerSettings);
 				});
 
+			services.AddSignalR(
+				options =>
+				{
+					options.AddFilter<AuthorizationContextHubFilter>();
+				})
+				.AddNewtonsoftJsonProtocol(options =>
+				{
+					ConfigureNewtonsoftJsonSerializerSettingsForApi(options.PayloadSerializerSettings);
+				});
+
+			services.AddHub<JobsHub, IJobsHub>();
+
 			if (postSetupServices.GeneralConfiguration.HostApiDocumentation)
 			{
 				string GetDocumentationFilePath(string assemblyLocation) => ioManager.ConcatPath(ioManager.GetDirectoryName(assemblyLocation), String.Concat(ioManager.GetFileNameWithoutExtension(assemblyLocation), ".xml"));
@@ -386,6 +402,7 @@ namespace Tgstation.Server.Host.Core
 			// configure root services
 			services.AddSingleton<IJobService, JobService>();
 			services.AddSingleton<IJobManager>(x => x.GetRequiredService<IJobService>());
+			services.AddSingleton<IPermissionsUpdateNotifyee, JobsHubGroupMapper>();
 
 			services.AddSingleton<InstanceManager>();
 			services.AddSingleton<IBridgeDispatcher>(x => x.GetRequiredService<InstanceManager>());
@@ -501,6 +518,9 @@ namespace Tgstation.Server.Host.Core
 				logger.LogTrace("Web control panel disabled!");
 #endif
 
+			// validate the API version
+			applicationBuilder.UseApiCompatibility();
+
 			// authenticate JWT tokens using our security pipeline if present, returns 401 if bad
 			applicationBuilder.UseAuthentication();
 
@@ -513,6 +533,17 @@ namespace Tgstation.Server.Host.Core
 			// setup endpoints
 			applicationBuilder.UseEndpoints(endpoints =>
 			{
+				// access to the signalR jobs hub
+				endpoints.MapHub<JobsHub>(
+					Routes.JobsHub,
+					options =>
+					{
+						options.Transports = HttpTransportType.ServerSentEvents;
+						options.CloseOnAuthenticationExpiration = true;
+					})
+					.RequireAuthorization()
+					.RequireCors(corsBuilder);
+
 				// majority of handling is done in the controllers
 				endpoints.MapControllers();
 			});
@@ -541,10 +572,18 @@ namespace Tgstation.Server.Host.Core
 			services.AddScoped<IApiHeadersProvider, ApiHeadersProvider>();
 			services.AddScoped<AuthenticationContextFactory>();
 			services.AddScoped<IAuthenticationContextFactory>(provider => provider.GetRequiredService<AuthenticationContextFactory>());
-			services.AddScoped(provider =>
-			{
-				return provider.GetRequiredService<AuthenticationContextFactory>().CurrentAuthenticationContext;
-			});
+
+			// what if you
+			// wanted to just do this:
+			// return provider.GetRequiredService<AuthenticationContextFactory>().CurrentAuthenticationContext
+			// But M$ said
+			// https://stackoverflow.com/questions/56792917/scoped-services-in-asp-net-core-with-signalr-hubs
+			services.AddScoped(provider => provider
+				.GetRequiredService<IHttpContextAccessor>()
+				.HttpContext
+				.RequestServices
+				.GetRequiredService<AuthenticationContextFactory>()
+				.CurrentAuthenticationContext);
 			services.AddScoped<IClaimsTransformation, AuthenticationContextClaimsTransformation>();
 			services.AddScoped<IAuthorizationFilter, AuthenticationContextAuthorizationFilter>();
 

src/Tgstation.Server.Host/Extensions/ApplicationBuilderExtensions.cs

@@ -119,6 +119,35 @@ namespace Tgstation.Server.Host.Extensions
 			});
 		}
 
+		/// <summary>
+		/// Check that the API version is the current major version if it's present in the headers.
+		/// </summary>
+		/// <param name="applicationBuilder">The <see cref="IApplicationBuilder"/> to configure.</param>
+		public static void UseApiCompatibility(this IApplicationBuilder applicationBuilder)
+		{
+			ArgumentNullException.ThrowIfNull(applicationBuilder);
+
+			applicationBuilder.Use(async (context, next) =>
+			{
+				var apiHeadersProvider = context.RequestServices.GetRequiredService<IApiHeadersProvider>();
+				if (apiHeadersProvider.ApiHeaders?.Compatible() == false)
+				{
+					await new JsonResult(
+						new ErrorMessageResponse(ErrorCode.ApiMismatch))
+					{
+						StatusCode = (int)HttpStatusCode.UpgradeRequired,
+					}
+					.ExecuteResultAsync(new ActionContext
+					{
+						HttpContext = context,
+					});
+					return;
+				}
+
+				await next();
+			});
+		}
+
 		/// <summary>
 		/// Add the X-Powered-By response header.
 		/// </summary>

src/Tgstation.Server.Host/Extensions/ServiceCollectionExtensions.cs

@@ -11,11 +11,13 @@ using Serilog;
 using Serilog.Configuration;
 using Serilog.Sinks.Elasticsearch;
 
+using Tgstation.Server.Api.Hubs;
 using Tgstation.Server.Host.Components.Chat.Providers;
 using Tgstation.Server.Host.Configuration;
 using Tgstation.Server.Host.IO;
 using Tgstation.Server.Host.Utils;
 using Tgstation.Server.Host.Utils.GitHub;
+using Tgstation.Server.Host.Utils.SignalR;
 
 namespace Tgstation.Server.Host.Extensions
 {
@@ -220,6 +222,23 @@ namespace Tgstation.Server.Host.Extensions
 			});
 		}
 
+		/// <summary>
+		/// Attempt to add the given <typeparamref name="THub"/> to services.
+		/// </summary>
+		/// <typeparam name="THub">The <see cref="Type"/> of the <see cref="Microsoft.AspNetCore.SignalR.Hub{T}"/> being added.</typeparam>
+		/// <typeparam name="THubMethods">The implementation <see cref="Type"/> of the <typeparamref name="THub"/>.</typeparam>
+		/// <param name="services">The <see cref="IServiceCollection"/> to add the <typeparamref name="THub"/> to.</param>
+		public static void AddHub<THub, THubMethods>(this IServiceCollection services)
+			where THub : ConnectionMappingHub<THub, THubMethods>
+			where THubMethods : class, IErrorHandlingHub
+		{
+			ArgumentNullException.ThrowIfNull(services);
+
+			services.TryAddSingleton(typeof(ComprehensiveHubContext<,>));
+			services.AddSingleton<IConnectionMappedHubContext<THub, THubMethods>>(provider => provider.GetRequiredService<ComprehensiveHubContext<THub, THubMethods>>());
+			services.AddSingleton<IHubConnectionMapper<THub, THubMethods>>(provider => provider.GetRequiredService<ComprehensiveHubContext<THub, THubMethods>>());
+		}
+
 		/// <summary>
 		/// Set the modifiable services to their default types.
 		/// </summary>

src/Tgstation.Server.Host/Jobs/JobService.cs

@@ -1,12 +1,17 @@
 using System;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 
+using Microsoft.AspNetCore.SignalR;
 using Microsoft.EntityFrameworkCore;
 using Microsoft.Extensions.Logging;
+
 using Serilog.Context;
+
+using Tgstation.Server.Api.Hubs;
 using Tgstation.Server.Api.Models.Response;
 using Tgstation.Server.Common.Extensions;
 using Tgstation.Server.Host.Components;
@@ -14,12 +19,23 @@ using Tgstation.Server.Host.Database;
 using Tgstation.Server.Host.Extensions;
 using Tgstation.Server.Host.Models;
 using Tgstation.Server.Host.Utils;
+using Tgstation.Server.Host.Utils.SignalR;
 
 namespace Tgstation.Server.Host.Jobs
 {
 	/// <inheritdoc cref="IJobService" />
 	sealed class JobService : IJobService, IDisposable
 	{
+		/// <summary>
+		/// The maximum rate at which hub clients can receive updates.
+		/// </summary>
+		const int MaxHubUpdatesPerSecond = 4;
+
+		/// <summary>
+		/// The <see cref="IHubContext"/> for the <see cref="JobsHub"/>.
+		/// </summary>
+		readonly IConnectionMappedHubContext<JobsHub, IJobsHub> hub;
+
 		/// <summary>
 		/// The <see cref="IServiceProvider"/> for the <see cref="JobService"/>.
 		/// </summary>
@@ -63,17 +79,21 @@ namespace Tgstation.Server.Host.Jobs
 		/// <summary>
 		/// Initializes a new instance of the <see cref="JobService"/> class.
 		/// </summary>
+		/// <param name="hub">The value of <see cref="hub"/>.</param>
 		/// <param name="databaseContextFactory">The value of <see cref="databaseContextFactory"/>.</param>
 		/// <param name="loggerFactory">The value of <see cref="loggerFactory"/>.</param>
 		/// <param name="logger">The value of <see cref="logger"/>.</param>
 		public JobService(
+			IConnectionMappedHubContext<JobsHub, IJobsHub> hub,
 			IDatabaseContextFactory databaseContextFactory,
 			ILoggerFactory loggerFactory,
 			ILogger<JobService> logger)
 		{
+			this.hub = hub ?? throw new ArgumentNullException(nameof(hub));
 			this.databaseContextFactory = databaseContextFactory ?? throw new ArgumentNullException(nameof(databaseContextFactory));
 			this.loggerFactory = loggerFactory ?? throw new ArgumentNullException(nameof(loggerFactory));
 			this.logger = logger ?? throw new ArgumentNullException(nameof(logger));
+
 			jobs = new Dictionary<long, JobHandler>();
 			activationTcs = new TaskCompletionSource<IInstanceCoreProvider>();
 			synchronizationLock = new object();
@@ -269,13 +289,12 @@ namespace Tgstation.Server.Host.Jobs
 			if (noMoreJobsShouldStart && !handler.Started)
 				await Extensions.TaskExtensions.InfiniteTask.WaitAsync(cancellationToken);
 
-			ValueTask<Job>? cancelTask = null;
+			var cancelTask = ValueTask.FromResult<Job>(null);
 			bool result;
 			using (jobCancellationToken.Register(() => cancelTask = CancelJob(job, canceller, true, cancellationToken)))
 				result = await handler.Wait(cancellationToken);
 
-			if (cancelTask.HasValue)
-				await cancelTask.Value;
+			await cancelTask;
 
 			return result;
 		}
@@ -292,23 +311,60 @@ namespace Tgstation.Server.Host.Jobs
 		/// <summary>
 		/// Runner for <see cref="JobHandler"/>s.
 		/// </summary>
-		/// <param name="job">The <see cref="Job"/> being run.</param>
+		/// <param name="job">The <see cref="Job"/> being run. Must be fully populated.</param>
 		/// <param name="operation">The <see cref="JobEntrypoint"/> for the <paramref name="job"/>.</param>
 		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
 		/// <returns>A <see cref="Task"/> representing the running operation.</returns>
+#pragma warning disable CA1506 // TODO: Decomplexify
 		async Task<bool> RunJob(Job job, JobEntrypoint operation, CancellationToken cancellationToken)
+#pragma warning restore CA1506
 		{
 			using (LogContext.PushProperty(SerilogContextHelper.JobIdContextProperty, job.Id))
 				try
 				{
 					void LogException(Exception ex) => logger.LogDebug(ex, "Job {jobId} exited with error!", job.Id);
 
+					var hubUpdatesTask = Task.CompletedTask;
 					var result = false;
+
+					Stopwatch stopwatch = null;
+					void QueueHubUpdate(JobResponse update)
+					{
+						var currentUpdatesTask = hubUpdatesTask;
+						async Task ChainHubUpdate()
+						{
+							await currentUpdatesTask;
+
+							// DCT: Cancellation token is for job, operation should always run
+							await hub
+								.Clients
+								.Group(JobsHub.HubGroupName(job))
+								.ReceiveJobUpdate(update, CancellationToken.None);
+						}
+
+						Stopwatch enteredLock = null;
+						try
+						{
+							if (stopwatch != null)
+							{
+								Monitor.Enter(stopwatch);
+								enteredLock = stopwatch;
+								if (stopwatch.ElapsedMilliseconds * MaxHubUpdatesPerSecond < 1)
+									return; // don't spam client
+							}
+
+							hubUpdatesTask = ChainHubUpdate();
+							stopwatch = Stopwatch.StartNew();
+						}
+						finally
+						{
+							if (enteredLock != null)
+								Monitor.Exit(enteredLock);
+						}
+					}
+
 					try
 					{
-						var oldJob = job;
-						job = new Job { Id = oldJob.Id };
-
 						void UpdateProgress(string stage, double? progress)
 						{
 							if (progress.HasValue
@@ -319,19 +375,26 @@ namespace Tgstation.Server.Host.Jobs
 								return;
 							}
 
+							int? newProgress = progress.HasValue ? (int)Math.Floor(progress.Value * 100) : null;
 							lock (synchronizationLock)
-								if (jobs.TryGetValue(oldJob.Id.Value, out var handler))
+								if (jobs.TryGetValue(job.Id.Value, out var handler))
 								{
 									handler.Stage = stage;
-									handler.Progress = progress.HasValue ? (int)Math.Floor(progress.Value * 100) : null;
+									handler.Progress = newProgress;
+
+									var updatedJob = job.ToApi();
+									updatedJob.Stage = stage;
+									updatedJob.Progress = newProgress;
+									QueueHubUpdate(updatedJob);
 								}
 						}
 
 						var instanceCoreProvider = await activationTcs.Task.WaitAsync(cancellationToken);
+						QueueHubUpdate(job.ToApi());
 
 						logger.LogTrace("Starting job...");
 						await operation(
-							instanceCoreProvider.GetInstance(oldJob.Instance),
+							instanceCoreProvider.GetInstance(job.Instance),
 							databaseContextFactory,
 							job,
 							new JobProgressReporter(
@@ -377,6 +440,31 @@ namespace Tgstation.Server.Host.Jobs
 						await databaseContext.Save(CancellationToken.None);
 					});
 
+					// Resetting the context here because I CBA to worry if the cache is being used
+					await databaseContextFactory.UseContext(async databaseContext =>
+					{
+						// Cancellation might be set in another async context, forced to reload here for the final hub update
+						// DCT: Cancellation token is for job, operation should always run
+						var finalJob = await databaseContext
+							.Jobs
+							.AsQueryable()
+							.Include(x => x.Instance)
+							.Include(x => x.StartedBy)
+							.Include(x => x.CancelledBy)
+							.Where(dbJob => dbJob.Id == job.Id.Value)
+							.FirstAsync(CancellationToken.None);
+						QueueHubUpdate(finalJob.ToApi());
+					});
+
+					try
+					{
+						await hubUpdatesTask;
+					}
+					catch (Exception ex)
+					{
+						logger.LogError(ex, "Error in hub updates chain task!");
+					}
+
 					return result;
 				}
 				finally

src/Tgstation.Server.Host/Jobs/JobsHub.cs

@@ -0,0 +1,52 @@
+using System;
+
+using Microsoft.AspNetCore.SignalR;
+
+using Tgstation.Server.Api.Hubs;
+using Tgstation.Server.Host.Models;
+using Tgstation.Server.Host.Security;
+using Tgstation.Server.Host.Utils.SignalR;
+
+namespace Tgstation.Server.Host.Jobs
+{
+	/// <summary>
+	/// A SignalR <see cref="Hub"/> for pushing job updates.
+	/// </summary>
+	sealed class JobsHub : ConnectionMappingHub<JobsHub, IJobsHub>
+	{
+		/// <summary>
+		/// Get the group name for a given <paramref name="instanceId"/>.
+		/// </summary>
+		/// <param name="instanceId">The <see cref="Instance"/> <see cref="Api.Models.EntityId.Id"/>.</param>
+		/// <returns>The name of the group for the <paramref name="instanceId"/>.</returns>
+		public static string HubGroupName(long instanceId)
+			=> $"instance-{instanceId}";
+
+		/// <summary>
+		/// Get the group name for a given <paramref name="job"/>.
+		/// </summary>
+		/// <param name="job">The <see cref="Job"/>.</param>
+		/// <returns>The name of the group for the <paramref name="job"/>.</returns>
+		public static string HubGroupName(Job job)
+		{
+			ArgumentNullException.ThrowIfNull(job);
+
+			if (job.Instance == null)
+				throw new InvalidOperationException("job.Instance was null!");
+
+			return HubGroupName(job.Instance.Id.Value);
+		}
+
+		/// <summary>
+		/// Initializes a new instance of the <see cref="JobsHub"/> class.
+		/// </summary>
+		/// <param name="connectionMapper">The <see cref="IHubConnectionMapper{THub, THubMethods}"/> for the <see cref="ConnectionMappingHub{TChildHub, THubMethods}"/>.</param>
+		/// <param name="authenticationContext">The <see cref="IAuthenticationContext"/> for the <see cref="ConnectionMappingHub{TChildHub, THubMethods}"/>.</param>
+		public JobsHub(
+			IHubConnectionMapper<JobsHub, IJobsHub> connectionMapper,
+			IAuthenticationContext authenticationContext)
+			: base(connectionMapper, authenticationContext)
+		{
+		}
+	}
+}

src/Tgstation.Server.Host/Jobs/JobsHubGroupMapper.cs

@@ -0,0 +1,172 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+using Microsoft.AspNetCore.SignalR;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Logging;
+
+using Tgstation.Server.Api.Hubs;
+using Tgstation.Server.Host.Database;
+using Tgstation.Server.Host.Models;
+using Tgstation.Server.Host.Security;
+using Tgstation.Server.Host.Utils.SignalR;
+
+namespace Tgstation.Server.Host.Jobs
+{
+	/// <summary>
+	/// Handles mapping groups for the <see cref="JobsHub"/>.
+	/// </summary>
+	sealed class JobsHubGroupMapper : IPermissionsUpdateNotifyee
+	{
+		/// <summary>
+		/// The <see cref="IHubContext"/> for the <see cref="JobsHub"/>.
+		/// </summary>
+		readonly IConnectionMappedHubContext<JobsHub, IJobsHub> hub;
+
+		/// <summary>
+		/// The <see cref="IServiceProvider"/> for the <see cref="JobService"/>.
+		/// </summary>
+		readonly IDatabaseContextFactory databaseContextFactory;
+
+		/// <summary>
+		/// The <see cref="ILogger"/> for the <see cref="JobService"/>.
+		/// </summary>
+		readonly ILogger<JobsHubGroupMapper> logger;
+
+		/// <summary>
+		/// Initializes a new instance of the <see cref="JobsHubGroupMapper"/> class.
+		/// </summary>
+		/// <param name="hub">The value of <see cref="hub"/>.</param>
+		/// <param name="databaseContextFactory">The value of <see cref="databaseContextFactory"/>.</param>
+		/// <param name="logger">The value of <see cref="logger"/>.</param>
+		public JobsHubGroupMapper(IConnectionMappedHubContext<JobsHub, IJobsHub> hub, IDatabaseContextFactory databaseContextFactory, ILogger<JobsHubGroupMapper> logger)
+		{
+			this.hub = hub ?? throw new ArgumentNullException(nameof(hub));
+			this.databaseContextFactory = databaseContextFactory ?? throw new ArgumentNullException(nameof(databaseContextFactory));
+			this.logger = logger ?? throw new ArgumentNullException(nameof(logger));
+
+			hub.OnConnectionMapGroups += MapConnectionGroups;
+		}
+
+		/// <inheritdoc />
+		public ValueTask InstancePermissionSetCreated(InstancePermissionSet instancePermissionSet, CancellationToken cancellationToken)
+		{
+			ArgumentNullException.ThrowIfNull(instancePermissionSet);
+			var permissionSetId = instancePermissionSet.PermissionSet.Id ?? instancePermissionSet.PermissionSetId;
+
+			logger.LogTrace("InstancePermissionSetCreated");
+			return RefreshHubGroups(
+				permissionSetId,
+				cancellationToken);
+		}
+
+		/// <inheritdoc />
+		public ValueTask UserDisabled(User user, CancellationToken cancellationToken)
+		{
+			ArgumentNullException.ThrowIfNull(user);
+			if (!user.Id.HasValue)
+				throw new InvalidOperationException("user.Id was null!");
+
+			logger.LogTrace("UserDisabled");
+			return hub.NotifyAndAbortUnauthedConnections(user, cancellationToken);
+		}
+
+		/// <inheritdoc />
+		public ValueTask InstancePermissionSetDeleted(PermissionSet permissionSet, CancellationToken cancellationToken)
+		{
+			ArgumentNullException.ThrowIfNull(permissionSet);
+			logger.LogTrace("InstancePermissionSetDeleted");
+			return RefreshHubGroups(
+				permissionSet.Id ?? throw new InvalidOperationException("permissionSet?.Id was null!"),
+				cancellationToken);
+		}
+
+		/// <summary>
+		/// Implementation of <see cref="IConnectionMappedHubContext{THub, THubMethods}.OnConnectionMapGroups"/>.
+		/// </summary>
+		/// <param name="authenticationContext">The <see cref="IAuthenticationContext"/> to map the groups for.</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>A <see cref="ValueTask{TResult}"/> resulting in an <see cref="IEnumerable{T}"/> of the <see cref="JobsHub"/> group names the user belongs in.</returns>
+		async ValueTask<IEnumerable<string>> MapConnectionGroups(IAuthenticationContext authenticationContext, CancellationToken cancellationToken)
+		{
+			ArgumentNullException.ThrowIfNull(authenticationContext);
+
+			List<long> permedInstanceIds = null;
+			await databaseContextFactory.UseContext(
+				async databaseContext =>
+					permedInstanceIds = await databaseContext
+						.InstancePermissionSets
+						.AsQueryable()
+						.Where(ips => ips.PermissionSetId == authenticationContext.PermissionSet.Id.Value)
+						.Select(ips => ips.Id)
+						.ToListAsync(cancellationToken));
+
+			return permedInstanceIds.Select(JobsHub.HubGroupName);
+		}
+
+		/// <summary>
+		/// Refresh the <see cref="hub"/> <see cref="Hub.Groups"/> for clients associated with a given <paramref name="permissionSetId"/>.
+		/// </summary>
+		/// <param name="permissionSetId">The <see cref="Api.Models.EntityId.Id"/> of the <see cref="PermissionSet"/> who's users need updating.</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>A <see cref="ValueTask"/> representing the running operation.</returns>
+		ValueTask RefreshHubGroups(long permissionSetId, CancellationToken cancellationToken)
+			=> databaseContextFactory.UseContext(
+				async databaseContext =>
+				{
+					logger.LogTrace("RefreshHubGroups");
+					var permissionSetUsers = await databaseContext
+						.Users
+						.Where(x => x.PermissionSet.Id == permissionSetId)
+						.ToListAsync(cancellationToken);
+					var allInstanceIds = await databaseContext
+						.Instances
+						.Select(
+							instance => instance.Id.Value)
+						.ToListAsync(cancellationToken);
+					var permissionSetAccessibleInstanceIds = await databaseContext
+						.InstancePermissionSets
+						.AsQueryable()
+						.Where(ips => ips.PermissionSetId == permissionSetId)
+						.Select(ips => ips.InstanceId)
+						.ToListAsync(cancellationToken);
+
+					var groupsToRemove = allInstanceIds
+						.Except(permissionSetAccessibleInstanceIds)
+						.Select(JobsHub.HubGroupName);
+
+					var groupsToAdd = permissionSetAccessibleInstanceIds
+						.Select(JobsHub.HubGroupName);
+
+					var connectionIds = permissionSetUsers
+						.SelectMany(user => hub.UserConnectionIds(user))
+						.ToList();
+
+					logger.LogTrace(
+						"Updating groups for the {connectionCount} hub connections of permission set {permissionSetId}. They may access {allowed}/{total} instances.",
+						connectionIds.Count,
+						permissionSetId,
+						permissionSetAccessibleInstanceIds.Count,
+						allInstanceIds.Count);
+
+					var removeTasks = connectionIds
+						.SelectMany(connectionId => groupsToRemove
+							.Select(groupName => hub
+								.Groups
+								.RemoveFromGroupAsync(connectionId, groupName, cancellationToken)));
+
+					var addTasks = connectionIds
+						.SelectMany(connectionId => groupsToAdd
+							.Select(groupName => hub
+								.Groups
+								.AddToGroupAsync(connectionId, groupName, cancellationToken)));
+
+					// Checked internally, the default implementations for these tasks complete synchronously
+					// https://github.com/dotnet/aspnetcore/blob/ce330d9d12f7676ff35c2223bd8a3b1e252a4e86/src/SignalR/server/Core/src/DefaultHubLifetimeManager.cs#L34-L70
+					await Task.WhenAll(removeTasks.Concat(addTasks));
+				});
+	}
+}

src/Tgstation.Server.Host/Security/AuthorizationContextHubFilter.cs

@@ -0,0 +1,89 @@
+using System;
+using System.Linq;
+using System.Threading.Tasks;
+
+using Microsoft.AspNetCore.SignalR;
+using Microsoft.Extensions.Logging;
+
+using Tgstation.Server.Api.Hubs;
+
+namespace Tgstation.Server.Host.Security
+{
+	/// <summary>
+	/// An <see cref="IHubFilter"/> that denies method calls and connections if the <see cref="IAuthenticationContext"/> is not valid for an authorized user.
+	/// </summary>
+	sealed class AuthorizationContextHubFilter : IHubFilter
+	{
+		/// <summary>
+		/// The <see cref="IAuthenticationContext"/> for the <see cref="AuthorizationContextHubFilter"/>.
+		/// </summary>
+		readonly IAuthenticationContext authenticationContext;
+
+		/// <summary>
+		/// The <see cref="ILogger"/> for the <see cref="AuthorizationContextHubFilter"/>.
+		/// </summary>
+		readonly ILogger<AuthorizationContextHubFilter> logger;
+
+		/// <summary>
+		/// Initializes a new instance of the <see cref="AuthorizationContextHubFilter"/> class.
+		/// </summary>
+		/// <param name="authenticationContext">The value of <see cref="authenticationContext"/>.</param>
+		/// <param name="logger">The value of <see cref="logger"/>.</param>
+		public AuthorizationContextHubFilter(
+			IAuthenticationContext authenticationContext,
+			ILogger<AuthorizationContextHubFilter> logger)
+		{
+			this.authenticationContext = authenticationContext ?? throw new ArgumentNullException(nameof(authenticationContext));
+			this.logger = logger ?? throw new ArgumentNullException(nameof(logger));
+		}
+
+		/// <inheritdoc />
+		public async Task OnConnectedAsync(HubLifetimeContext context, Func<HubLifetimeContext, Task> next)
+		{
+			ArgumentNullException.ThrowIfNull(context);
+			if (await ValidateAuthenticationContext(context.Hub))
+				await next(context);
+		}
+
+		/// <inheritdoc />
+		public async ValueTask<object> InvokeMethodAsync(HubInvocationContext invocationContext, Func<HubInvocationContext, ValueTask<object>> next)
+		{
+			ArgumentNullException.ThrowIfNull(invocationContext);
+			if (await ValidateAuthenticationContext(invocationContext.Hub))
+				return await next(invocationContext);
+
+			return null;
+		}
+
+		/// <summary>
+		/// Validates the <see cref="IAuthenticationContext"/> for the hub event.
+		/// </summary>
+		/// <param name="hub">The current <see cref="Hub"/>.</param>
+		/// <returns><see langword="true"/> if the hub call should continue, <see langword="false"/> if it shouldn't and has been aborted.</returns>
+		async ValueTask<bool> ValidateAuthenticationContext(Hub hub)
+		{
+			if (!authenticationContext.Valid)
+				logger.LogTrace("The token for connection {connectionId} is no longer authenticated! Aborting...", hub.Context.ConnectionId);
+			else if (!authenticationContext.User.Enabled.Value)
+				logger.LogTrace("The token for connection {connectionId} is no longer authorized! Aborting...", hub.Context.ConnectionId);
+			else
+				return true;
+
+			var hubType = hub.GetType();
+			var allHubProperties = hubType.GetProperties();
+			var typedClientsProperty = allHubProperties.Single(
+				prop => prop.PropertyType.IsConstructedGenericType
+					&& prop.Name == nameof(hub.Clients));
+			var clients = typedClientsProperty.GetValue(hub);
+			var callerProperty = clients.GetType().GetProperty(nameof(hub.Clients.Caller));
+			var caller = callerProperty.GetValue(clients);
+
+			if (caller is not IErrorHandlingHub specifiedHub)
+				throw new InvalidOperationException("This filter only supports IErrorHandlingHubs");
+
+			await specifiedHub.AbortingConnection(ConnectionAbortReason.TokenInvalid, hub.Context.ConnectionAborted);
+			hub.Context.Abort();
+			return false;
+		}
+	}
+}

src/Tgstation.Server.Host/Security/IPermissionsUpdateNotifyee.cs

@@ -0,0 +1,37 @@
+using System.Threading;
+using System.Threading.Tasks;
+
+using Tgstation.Server.Host.Models;
+
+namespace Tgstation.Server.Host.Security
+{
+	/// <summary>
+	/// Receives notifications about permissions updates.
+	/// </summary>
+	public interface IPermissionsUpdateNotifyee
+	{
+		/// <summary>
+		/// Called when a given <paramref name="instancePermissionSet"/> is successfully created.
+		/// </summary>
+		/// <param name="instancePermissionSet">The <see cref="InstancePermissionSet"/>. <see cref="InstancePermissionSet.PermissionSet"/> must be populated.</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>A <see cref="ValueTask"/> representing the running operation.</returns>
+		ValueTask InstancePermissionSetCreated(InstancePermissionSet instancePermissionSet, CancellationToken cancellationToken);
+
+		/// <summary>
+		/// Called when an <see cref="InstancePermissionSet"/> is successfully deleted.
+		/// </summary>
+		/// <param name="permissionSet">The <see cref="PermissionSet"/> of the deleted <see cref="InstancePermissionSet"/>.</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>A <see cref="ValueTask"/> representing the running operation.</returns>
+		ValueTask InstancePermissionSetDeleted(PermissionSet permissionSet, CancellationToken cancellationToken);
+
+		/// <summary>
+		/// Called when a given <see cref="User"/> is successfully disabled.
+		/// </summary>
+		/// <param name="user">The <see cref="User"/> that was disabled.</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>A <see cref="ValueTask"/> representing the running operation.</returns>
+		ValueTask UserDisabled(User user, CancellationToken cancellationToken);
+	}
+}

src/Tgstation.Server.Host/Tgstation.Server.Host.csproj

@@ -78,6 +78,8 @@
     <PackageReference Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="6.0.24" />
     <!-- Usage: Support ""legacy"" Newotonsoft.Json in HTTP pipeline. The rest of our codebase uses Newtonsoft. -->
     <PackageReference Include="Microsoft.AspNetCore.Mvc.NewtonsoftJson" Version="6.0.24" />
+    <!-- Usage: Using target JSON serializer for API -->
+    <PackageReference Include="Microsoft.AspNetCore.SignalR.Protocols.NewtonsoftJson" Version="7.0.13" />
     <!-- Usage: Database ORM -->
     <PackageReference Include="Microsoft.EntityFrameworkCore" Version="7.0.13" />
     <!-- Usage: Automatic migration generation using command line -->

src/Tgstation.Server.Host/Utils/ApiHeadersProvider.cs

@@ -1,4 +1,4 @@
-using System;
+using System;
 
 using Microsoft.AspNetCore.Http;
 
@@ -12,7 +12,7 @@ namespace Tgstation.Server.Host.Utils
 		/// <inheritdoc />
 		public ApiHeaders ApiHeaders => attemptedApiHeadersCreation
 			? apiHeaders
-			: CreateApiHeaders(true);
+			: CreateApiHeaders(false);
 
 		/// <inheritdoc />
 		public HeadersException HeadersException { get; private set; }
@@ -42,33 +42,31 @@ namespace Tgstation.Server.Host.Utils
 		}
 
 		/// <inheritdoc />
-		public ApiHeaders CreateAuthlessHeaders() => CreateApiHeaders(false);
+		public ApiHeaders CreateAuthlessHeaders() => CreateApiHeaders(true);
 
 		/// <summary>
 		/// Attempt to parse <see cref="Api.ApiHeaders"/> from the <see cref="HttpContext"/>, optionally populating the <see langword="class"/> properties.
 		/// </summary>
-		/// <param name="includeAuthAndSetProperties">If the <see cref="HeaderErrorTypes.AuthorizationMissing"/> error should be ignored and <see cref="ApiHeaders"/>/<see cref="HeadersException"/> should be populated.</param>
-		/// <returns>A newly parsed <see cref="Api.ApiHeaders"/> <see langword="class"/> or <see langword="null"/> if <paramref name="includeAuthAndSetProperties"/> was set and the parse failed.</returns>
-		ApiHeaders CreateApiHeaders(bool includeAuthAndSetProperties)
+		/// <param name="authless">If the <see cref="HeaderErrorTypes.AuthorizationMissing"/> error should be ignored and <see cref="ApiHeaders"/>/<see cref="HeadersException"/> should not be populated.</param>
+		/// <returns>A newly parsed <see cref="Api.ApiHeaders"/> <see langword="class"/> or <see langword="null"/> if <paramref name="authless"/> was set and the parse failed.</returns>
+		ApiHeaders CreateApiHeaders(bool authless)
 		{
 			if (httpContextAccessor.HttpContext == null)
 				throw new InvalidOperationException("httpContextAccessor has no HttpContext!");
 
-			var request = httpContextAccessor.HttpContext.Request;
-			var ignoreMissingAuth = !includeAuthAndSetProperties;
-
-			if (includeAuthAndSetProperties)
+			var typedHeaders = httpContextAccessor.HttpContext.Request.GetTypedHeaders();
+			if (!authless)
 				attemptedApiHeadersCreation = true;
 
 			try
 			{
-				var headers = new ApiHeaders(request.GetTypedHeaders(), ignoreMissingAuth);
-				if (includeAuthAndSetProperties)
+				var headers = new ApiHeaders(typedHeaders, authless, !authless);
+				if (!authless)
 					apiHeaders = headers;
 
 				return headers;
 			}
-			catch (HeadersException ex) when (includeAuthAndSetProperties)
+			catch (HeadersException ex) when (!authless)
 			{
 				HeadersException = ex;
 				return null;

src/Tgstation.Server.Host/Utils/SignalR/ComprehensiveHubContext.cs

@@ -0,0 +1,165 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+using Microsoft.AspNetCore.SignalR;
+using Microsoft.Extensions.Logging;
+
+using Tgstation.Server.Api.Hubs;
+using Tgstation.Server.Common.Extensions;
+using Tgstation.Server.Host.Core;
+using Tgstation.Server.Host.Models;
+using Tgstation.Server.Host.Security;
+
+namespace Tgstation.Server.Host.Utils.SignalR
+{
+	/// <summary>
+	/// An implementation of <see cref="IHubContext{THub}"/> with <see cref="User"/> connection ID mapping.
+	/// </summary>
+	/// <typeparam name="THub">The <see cref="Hub"/> the <see cref="ComprehensiveHubContext{THub, THubMethods}"/> is for.</typeparam>
+	/// <typeparam name="THubMethods">The interface <see cref="IErrorHandlingHub"/> for implementing <see cref="Hub{T}"/> methods.</typeparam>
+	sealed class ComprehensiveHubContext<THub, THubMethods> : IConnectionMappedHubContext<THub, THubMethods>, IHubConnectionMapper<THub, THubMethods>, IRestartHandler
+		where THub : ConnectionMappingHub<THub, THubMethods>
+		where THubMethods : class, IErrorHandlingHub
+	{
+		/// <inheritdoc />
+		public IHubClients<THubMethods> Clients => wrappedHubContext.Clients;
+
+		/// <inheritdoc />
+		public IGroupManager Groups => wrappedHubContext.Groups;
+
+		/// <summary>
+		/// The <see cref="IHubContext{THub}"/> being wrapped.
+		/// </summary>
+		readonly IHubContext<THub, THubMethods> wrappedHubContext;
+
+		/// <summary>
+		/// The <see cref="ILogger"/> for the <see cref="ComprehensiveHubContext{THub, THubMethods}"/>.
+		/// </summary>
+		readonly ILogger<ComprehensiveHubContext<THub, THubMethods>> logger;
+
+		/// <summary>
+		/// Map of <see cref="User"/> <see cref="Api.Models.EntityId.Id"/>s to their associated <see cref="HubCallerContext"/>s.
+		/// </summary>
+		readonly ConcurrentDictionary<long, Dictionary<string, HubCallerContext>> userConnections;
+
+		/// <inheritdoc />
+		public event Func<IAuthenticationContext, CancellationToken, ValueTask<IEnumerable<string>>> OnConnectionMapGroups;
+
+		/// <summary>
+		/// Initializes a new instance of the <see cref="ComprehensiveHubContext{THub, THubMethods}"/> class.
+		/// </summary>
+		/// <param name="wrappedHubContext">The value of <see cref="wrappedHubContext"/>.</param>
+		/// <param name="serverControl">The <see cref="IServerControl"/> to <see cref="IServerControl.RegisterForRestart(IRestartHandler)"/> with.</param>
+		/// <param name="logger">The value of <see cref="logger"/>.</param>
+		public ComprehensiveHubContext(
+			IHubContext<THub, THubMethods> wrappedHubContext,
+			IServerControl serverControl,
+			ILogger<ComprehensiveHubContext<THub, THubMethods>> logger)
+		{
+			this.wrappedHubContext = wrappedHubContext ?? throw new ArgumentNullException(nameof(wrappedHubContext));
+			ArgumentNullException.ThrowIfNull(serverControl);
+			this.logger = logger ?? throw new ArgumentNullException(nameof(logger));
+
+			userConnections = new ConcurrentDictionary<long, Dictionary<string, HubCallerContext>>();
+
+			serverControl.RegisterForRestart(this);
+		}
+
+		/// <inheritdoc />
+		public List<string> UserConnectionIds(User user)
+		{
+			ArgumentNullException.ThrowIfNull(user);
+			var connectionIds = userConnections.GetOrAdd(user.Id.Value, _ => new Dictionary<string, HubCallerContext>());
+			lock (connectionIds)
+				return connectionIds.Keys.ToList();
+		}
+
+		/// <inheritdoc />
+		public async ValueTask UserConnected(IAuthenticationContext authenticationContext, THub hub, CancellationToken cancellationToken)
+		{
+			ArgumentNullException.ThrowIfNull(authenticationContext);
+			ArgumentNullException.ThrowIfNull(hub);
+
+			var userId = authenticationContext.User.Id.Value;
+			var context = hub.Context;
+			logger.LogTrace(
+				"Mapping user {userId} to hub connection ID: {connectionId}",
+				userId,
+				context.ConnectionId);
+
+			var mappedGroupsTask = OnConnectionMapGroups(authenticationContext, cancellationToken);
+			userConnections.AddOrUpdate(
+				userId,
+				_ => new Dictionary<string, HubCallerContext>
+				{
+					{ context.ConnectionId, context },
+				},
+				(_, old) =>
+				{
+					lock (old)
+						old[context.ConnectionId] = context;
+
+					return old;
+				});
+
+			var mappedGroups = await mappedGroupsTask;
+			await Task.WhenAll(
+				mappedGroups.Select(
+					group => hub.Groups.AddToGroupAsync(context.ConnectionId, group, cancellationToken)));
+		}
+
+		/// <inheritdoc />
+		public void UserDisconnected(string connectionId)
+		{
+			ArgumentNullException.ThrowIfNull(connectionId);
+			foreach (var kvp in userConnections)
+				lock (kvp.Value)
+					if (kvp.Value.Remove(connectionId))
+						logger.LogTrace("User {userId} disconnected connection ID: {connectionId}", kvp.Key, connectionId);
+		}
+
+		/// <inheritdoc />
+		public ValueTask NotifyAndAbortUnauthedConnections(User user, CancellationToken cancellationToken)
+		{
+			ArgumentNullException.ThrowIfNull(user);
+			logger.LogTrace("NotifyAndAbortUnauthedConnections. UID {userId}", user.Id.Value);
+
+			List<HubCallerContext> connections = null;
+			userConnections.AddOrUpdate(
+				user.Id.Value,
+				_ => new Dictionary<string, HubCallerContext>(),
+				(_, old) =>
+				{
+					lock (old)
+					{
+						connections = old.Values.ToList();
+						old.Clear();
+					}
+
+					return old;
+				});
+
+			async ValueTask NotifyAndAbortConnection(HubCallerContext context)
+			{
+				await Clients
+					.Client(context.ConnectionId)
+					.AbortingConnection(ConnectionAbortReason.TokenInvalid, cancellationToken);
+				context.Abort();
+			}
+
+			return ValueTaskExtensions.WhenAll(connections.Select(NotifyAndAbortConnection));
+		}
+
+		/// <inheritdoc />
+		public async ValueTask HandleRestart(Version updateVersion, bool handlerMayDelayShutdownWithExtremelyLongRunningTasks, CancellationToken cancellationToken)
+		{
+			logger.LogTrace("HandleRestart. {connectionCount} active connections", userConnections.Count);
+			await Clients.All.AbortingConnection(ConnectionAbortReason.ServerRestart, cancellationToken);
+			userConnections.Clear();
+		}
+	}
+}

src/Tgstation.Server.Host/Utils/SignalR/ConnectionMappingHub.cs

@@ -0,0 +1,60 @@
+using System;
+using System.Threading.Tasks;
+
+using Microsoft.AspNetCore.Authorization;
+using Microsoft.AspNetCore.SignalR;
+
+using Tgstation.Server.Api.Hubs;
+using Tgstation.Server.Host.Security;
+
+namespace Tgstation.Server.Host.Utils.SignalR
+{
+	/// <summary>
+	/// Base <see langword="class"/> for <see cref="Hub{T}"/>s that want to map their connection IDs to <see cref="Models.PermissionSet"/>s.
+	/// </summary>
+	/// <typeparam name="TChildHub">The child <see langword="class"/> inheriting from the <see cref="ConnectionMappingHub{TChildHub, THubMethods}"/>.</typeparam>
+	/// <typeparam name="THubMethods">The interface <see cref="IErrorHandlingHub"/> for implementing <see cref="Hub{T}"/> methods.</typeparam>
+	[TgsAuthorize]
+	abstract class ConnectionMappingHub<TChildHub, THubMethods> : Hub<THubMethods>
+		where TChildHub : ConnectionMappingHub<TChildHub, THubMethods>
+		where THubMethods : class, IErrorHandlingHub
+	{
+		/// <summary>
+		/// The <see cref="IHubConnectionMapper{THub, THubMethods}"/> used to map connections.
+		/// </summary>
+		readonly IHubConnectionMapper<TChildHub, THubMethods> connectionMapper;
+
+		/// <summary>
+		/// The <see cref="IAuthenticationContext"/> for the <see cref="ConnectionMappingHub{TChildHub, THubMethods}"/>.
+		/// </summary>
+		readonly IAuthenticationContext authenticationContext;
+
+		/// <summary>
+		/// Initializes a new instance of the <see cref="ConnectionMappingHub{TParentHub, TClientMethods}"/> class.
+		/// </summary>
+		/// <param name="connectionMapper">The value of <see cref="connectionMapper"/>.</param>
+		/// <param name="authenticationContext">The value of <see cref="authenticationContext"/>.</param>
+		protected ConnectionMappingHub(
+			IHubConnectionMapper<TChildHub, THubMethods> connectionMapper,
+			IAuthenticationContext authenticationContext)
+		{
+			this.connectionMapper = connectionMapper ?? throw new ArgumentNullException(nameof(connectionMapper));
+			this.authenticationContext = authenticationContext ?? throw new ArgumentNullException(nameof(authenticationContext));
+		}
+
+		/// <inheritdoc />
+		public override async Task OnConnectedAsync()
+		{
+			await connectionMapper.UserConnected(authenticationContext, (TChildHub)this, Context.ConnectionAborted);
+			await base.OnConnectedAsync();
+		}
+
+		/// <inheritdoc />
+		[AllowAnonymous]
+		public override Task OnDisconnectedAsync(Exception exception)
+		{
+			connectionMapper.UserDisconnected(Context.ConnectionId);
+			return base.OnDisconnectedAsync(exception);
+		}
+	}
+}

src/Tgstation.Server.Host/Utils/SignalR/IConnectionMappedHubContext.cs

@@ -0,0 +1,43 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+using Microsoft.AspNetCore.SignalR;
+
+using Tgstation.Server.Api.Hubs;
+using Tgstation.Server.Host.Models;
+using Tgstation.Server.Host.Security;
+
+namespace Tgstation.Server.Host.Utils.SignalR
+{
+	/// <summary>
+	/// A <see cref="IHubContext{THub}"/> that maps <see cref="User"/>s to their connection IDs.
+	/// </summary>
+	/// <typeparam name="THub">The <see cref="Hub"/> the <see cref="IConnectionMappedHubContext{THub, THubMethods}"/> is for.</typeparam>
+	/// <typeparam name="THubMethods">The interface <see langword="class"/> for implementing <see cref="Hub{T}"/> methods.</typeparam>
+	interface IConnectionMappedHubContext<THub, THubMethods> : IHubContext<THub, THubMethods>
+		where THub : Hub<THubMethods>
+		where THubMethods : class, IErrorHandlingHub
+	{
+		/// <summary>
+		/// Called when a user connects. Should return an <see cref="IEnumerable{T}"/> of hub group names the given <see cref="IAuthenticationContext"/> belongs in.
+		/// </summary>
+		event Func<IAuthenticationContext, CancellationToken, ValueTask<IEnumerable<string>>> OnConnectionMapGroups;
+
+		/// <summary>
+		/// Gets a <see cref="List{T}"/> of current connection IDs for a given <paramref name="user"/>.
+		/// </summary>
+		/// <param name="user">The <see cref="User"/> to get connection IDs for.</param>
+		/// <returns>A <see cref="List{T}"/> representing the active connection IDs of the <paramref name="user"/>.</returns>
+		List<string> UserConnectionIds(User user);
+
+		/// <summary>
+		/// Calls <see cref="IErrorHandlingHub.AbortingConnection(ConnectionAbortReason, CancellationToken)"/> with <see cref="ConnectionAbortReason.TokenInvalid"/> on and aborts the connections associated with the given <paramref name="user"/>.
+		/// </summary>
+		/// <param name="user">The <see cref="User"/> to abort the connections of.</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>A <see cref="ValueTask"/> representing the running operation.</returns>
+		ValueTask NotifyAndAbortUnauthedConnections(User user, CancellationToken cancellationToken);
+	}
+}

src/Tgstation.Server.Host/Utils/SignalR/IHubConnectionMapper.cs

@@ -0,0 +1,36 @@
+using System.Threading;
+using System.Threading.Tasks;
+
+using Microsoft.AspNetCore.SignalR;
+
+using Tgstation.Server.Api.Hubs;
+using Tgstation.Server.Host.Models;
+using Tgstation.Server.Host.Security;
+
+namespace Tgstation.Server.Host.Utils.SignalR
+{
+	/// <summary>
+	/// Handles mapping connection IDs to <see cref="User"/>s for a given <typeparamref name="THub"/>.
+	/// </summary>
+	/// <typeparam name="THub">The <see cref="Hub"/> whose connections are being mapped.</typeparam>
+	/// <typeparam name="THubMethods">The interface <see cref="IErrorHandlingHub"/> for implementing <see cref="Hub{T}"/> methods.</typeparam>
+	interface IHubConnectionMapper<THub, THubMethods>
+		where THub : ConnectionMappingHub<THub, THubMethods>
+		where THubMethods : class, IErrorHandlingHub
+	{
+		/// <summary>
+		/// To be called when a hub connection is made.
+		/// </summary>
+		/// <param name="authenticationContext">The <see cref="IAuthenticationContext"/> associated with the connection.</param>
+		/// <param name="hub">The <typeparamref name="THub"/>.</param>
+		/// <param name="cancellationToken">The <see cref="CancellationToken"/> for the operation.</param>
+		/// <returns>A <see cref="ValueTask"/> representing the running operation.</returns>
+		ValueTask UserConnected(IAuthenticationContext authenticationContext, THub hub, CancellationToken cancellationToken);
+
+		/// <summary>
+		/// To be called when a hub connection is terminated.
+		/// </summary>
+		/// <param name="connectionId">The connection ID.</param>
+		void UserDisconnected(string connectionId);
+	}
+}

tests/Tgstation.Server.Api.Tests/TestApiHeaders.cs

@@ -43,7 +43,7 @@ namespace Tgstation.Server.Api.Tests
 					{ "User-Agent", userAgent }
 				};
 
-				return new ApiHeaders(new RequestHeaders(headers), false);
+				return new ApiHeaders(new RequestHeaders(headers), false, false);
 			};
 
 			var header = TestHeader(BrowserHeader);

tests/Tgstation.Server.Client.Tests/Tgstation.Server.Client.Tests.csproj

@@ -5,6 +5,10 @@
     <TargetFramework>$(TgsFrameworkVersion)</TargetFramework>
   </PropertyGroup>
 
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="7.0.0" />
+  </ItemGroup>
+
   <ItemGroup>
     <ProjectReference Include="..\..\src\Tgstation.Server.Client\Tgstation.Server.Client.csproj" />
   </ItemGroup>

tests/Tgstation.Server.Tests/Live/Instance/JobsHubTests.cs

@@ -0,0 +1,293 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+using Microsoft.AspNetCore.SignalR.Client;
+using Microsoft.VisualStudio.TestTools.UnitTesting;
+
+using Tgstation.Server.Api.Hubs;
+using Tgstation.Server.Api.Models.Request;
+using Tgstation.Server.Api.Models.Response;
+using Tgstation.Server.Client;
+using Tgstation.Server.Common.Extensions;
+
+namespace Tgstation.Server.Tests.Live.Instance
+{
+	sealed class JobsHubTests : IJobsHub
+	{
+		const int ActiveConnections = 2;
+
+		readonly IServerClient permedUser;
+		readonly IServerClient permlessUser;
+
+		readonly TaskCompletionSource finishTcs;
+
+		readonly ConcurrentDictionary<long, JobResponse> seenJobs;
+
+		readonly HashSet<long> permlessSeenJobs;
+
+		HubConnection conn1, conn2;
+		int expectedReboots;
+		bool permlessIsPermed;
+
+		long? permlessPsId;
+
+		public JobsHubTests(IServerClient permedUser, IServerClient permlessUser)
+		{
+			this.permedUser = permedUser;
+			this.permlessUser = permlessUser;
+
+			Assert.AreNotSame(permedUser, permlessUser);
+
+			finishTcs = new TaskCompletionSource();
+
+			seenJobs = new ConcurrentDictionary<long, JobResponse>();
+			permlessSeenJobs = new HashSet<long>();
+		}
+
+		public Task ReceiveJobUpdate(JobResponse job, CancellationToken cancellationToken)
+		{
+			try
+			{
+				Assert.IsTrue(job.InstanceId.HasValue);
+				Assert.IsNotNull(job.StartedBy);
+				Assert.IsTrue(job.StartedBy.Id.HasValue);
+				Assert.IsTrue(job.StartedAt.HasValue);
+				Assert.IsNotNull(job.Description);
+
+				seenJobs.AddOrUpdate(job.Id.Value, job, (_, old) =>
+				{
+					Assert.IsFalse(old.StoppedAt.HasValue, $"Received update for job {job.Id} after it had completed!");
+
+					return job;
+				});
+			}
+			catch(Exception ex)
+			{
+				finishTcs.SetException(ex);
+			}
+
+			return Task.CompletedTask;
+		}
+
+
+		class ShouldNeverReceiveUpdates : IJobsHub
+		{
+			public Action<JobResponse> Callback { get; set; }
+			public Func<ConnectionAbortReason, CancellationToken, Task> Error { get; set; }
+
+			public Task AbortingConnection(ConnectionAbortReason reason, CancellationToken cancellationToken)
+				=> Error(reason, cancellationToken);
+
+			public Task ReceiveJobUpdate(JobResponse job, CancellationToken cancellationToken)
+			{
+				Callback(job);
+				return Task.CompletedTask;
+			}
+		}
+
+		public async Task Run(CancellationToken cancellationToken)
+		{
+			var neverReceiverTcs = new TaskCompletionSource();
+			var neverReceiver = new ShouldNeverReceiveUpdates()
+			{
+				Callback = job =>
+				{
+					if (!permlessIsPermed)
+						neverReceiverTcs.TrySetException(new Exception($"ShouldNeverReceiveUpdates received an update for job {job.Id}!"));
+					else
+						lock (permlessSeenJobs)
+							permlessSeenJobs.Add(job.Id.Value);
+				},
+				Error = AbortingConnection,
+			};
+
+			await using (conn1 = (HubConnection)await permedUser.SubscribeToJobUpdates(
+				this,
+				null,
+				null,
+				cancellationToken))
+			await using (conn2 = (HubConnection)await permlessUser.SubscribeToJobUpdates(
+				neverReceiver,
+				null,
+				null,
+				cancellationToken))
+			{
+				Console.WriteLine($"Initial conn1: {conn1.ConnectionId}");
+				Console.WriteLine($"Initial conn2: {conn2.ConnectionId}");
+
+				conn1.Reconnected += (newId) =>
+				{
+					Console.WriteLine($"conn1 reconnected: {newId}");
+					return Task.CompletedTask;
+				};
+				conn2.Reconnected += (newId) =>
+				{
+					Console.WriteLine($"conn1 reconnected: {newId}");
+					return Task.CompletedTask;
+				};
+
+				var completedTask = await Task.WhenAny(finishTcs.Task, neverReceiverTcs.Task);
+				await completedTask;
+			}
+
+			neverReceiverTcs.TrySetResult();
+			await neverReceiverTcs.Task;
+
+			var allInstances = await permedUser.Instances.List(null, cancellationToken);
+
+			async ValueTask<List<JobResponse>> CheckInstance(InstanceResponse instance)
+			{
+				var wasOffline = !instance.Online.Value;
+				if (wasOffline)
+					await permedUser.Instances.Update(new InstanceUpdateRequest
+					{
+						Id = instance.Id,
+						Online = true,
+					}, cancellationToken);
+
+				var jobs = await permedUser.Instances.CreateClient(instance).Jobs.List(null, cancellationToken);
+				if (wasOffline)
+					await permedUser.Instances.Update(new InstanceUpdateRequest
+					{
+						Id = instance.Id,
+						Online = false,
+					}, cancellationToken);
+
+				return jobs;
+			}
+
+			var allJobsTask = allInstances
+				.Select(CheckInstance);
+
+			var allJobs = (await ValueTaskExtensions.WhenAll(allJobsTask, allInstances.Count)).SelectMany(x => x).ToList();
+			var missableMissedJobs = 0;
+			foreach (var job in allJobs)
+			{
+				var seenThisJob = seenJobs.TryGetValue(job.Id.Value, out var hubJob);
+				if (seenThisJob)
+				{
+					Assert.AreEqual(job.StoppedAt, hubJob.StoppedAt);
+					Assert.AreEqual(job.InstanceId, hubJob.InstanceId);
+					Assert.AreEqual(job.ExceptionDetails, hubJob.ExceptionDetails);
+					Assert.AreEqual(job.Stage, hubJob.Stage);
+					Assert.AreEqual(job.CancelledBy?.Id, hubJob.CancelledBy?.Id);
+					Assert.AreEqual(job.Cancelled, hubJob.Cancelled);
+					Assert.AreEqual(job.StartedBy?.Id, hubJob.StartedBy?.Id);
+					Assert.AreEqual(job.CancelRight, hubJob.CancelRight);
+					Assert.AreEqual(job.CancelRightsType, hubJob.CancelRightsType);
+					Assert.AreEqual(job.Progress, hubJob.Progress);
+					Assert.AreEqual(job.Description, hubJob.Description);
+					Assert.AreEqual(job.ErrorCode, hubJob.ErrorCode);
+					Assert.AreEqual(job.StartedAt, hubJob.StartedAt);
+				}
+				else
+				{
+					var wasMissableJob = job.Description.StartsWith("Reconnect chat bot")
+						|| job.Description.StartsWith("Instance startup watchdog reattach")
+						|| job.Description.StartsWith("Instance startup watchdog launch");
+					Assert.IsTrue(wasMissableJob);
+					++missableMissedJobs;
+				}
+			}
+
+			// some instances may be detached, but our cache remains
+			var accountedJobs = allJobs.Count - missableMissedJobs;
+			var accountedSeenJobs = seenJobs.Where(x => allInstances.Any(i => i.Id.Value == x.Value.InstanceId)).Count();
+			Assert.AreEqual(accountedJobs, accountedSeenJobs);
+			Assert.IsTrue(accountedJobs <= seenJobs.Count);
+			Assert.AreNotEqual(0, permlessSeenJobs.Count);
+			Assert.IsTrue(permlessSeenJobs.Count < seenJobs.Count);
+			Assert.IsTrue(permlessSeenJobs.All(id => seenJobs.ContainsKey(id)));
+
+			await using var conn3 = (HubConnection)await permedUser.SubscribeToJobUpdates(
+				this,
+				null,
+				null,
+				cancellationToken);
+
+			Assert.AreEqual(HubConnectionState.Connected, conn3.State);
+			await permlessUser.DisposeAsync();
+			await permedUser.DisposeAsync();
+			Assert.AreEqual(0, expectedReboots);
+		}
+
+		public void ExpectShutdown()
+		{
+			Assert.AreEqual(0, Interlocked.Exchange(ref expectedReboots, ActiveConnections));
+			Assert.AreEqual(HubConnectionState.Connected, conn1.State);
+			Assert.AreEqual(HubConnectionState.Connected, conn2.State);
+		}
+
+		public async ValueTask WaitForReconnect(CancellationToken cancellationToken)
+		{
+			Assert.AreEqual(0, expectedReboots);
+			await Task.WhenAll(conn1.StopAsync(cancellationToken), conn2.StopAsync(cancellationToken));
+
+			Assert.AreEqual(HubConnectionState.Disconnected, conn1.State);
+			Assert.AreEqual(HubConnectionState.Disconnected, conn2.State);
+
+			// force token refreshs
+			await Task.WhenAll(permedUser.Administration.Read(cancellationToken).AsTask(), permlessUser.Instances.List(null, cancellationToken).AsTask());
+
+			await Task.WhenAll(conn1.StartAsync(cancellationToken), conn2.StartAsync(cancellationToken));
+
+			Assert.AreEqual(HubConnectionState.Connected, conn1.State);
+			Assert.AreEqual(HubConnectionState.Connected, conn2.State);
+			Console.WriteLine($"New conn1: {conn1.ConnectionId}");
+			Console.WriteLine($"New conn2: {conn2.ConnectionId}");
+
+			if (!permlessPsId.HasValue)
+			{
+				var permlessUserId = long.Parse(permlessUser.Token.ParseJwt().Subject);
+				permlessPsId = (await permedUser.Users.GetId(new Api.Models.EntityId
+				{
+					Id = permlessUserId
+				}, cancellationToken)).PermissionSet.Id;
+			}
+
+			var instancesTask = permedUser.Instances.List(null, cancellationToken);
+
+			permlessIsPermed = !permlessIsPermed;
+
+			var instances = await instancesTask;
+			await ValueTaskExtensions.WhenAll(
+				instances
+				.Where(instance => instance.Online.Value)
+				.Select<InstanceResponse, ValueTask>(async instance =>
+				{
+					var ic = permedUser.Instances.CreateClient(instance);
+					if (permlessIsPermed)
+						await ic.PermissionSets.Create(new InstancePermissionSetRequest
+						{
+							PermissionSetId = permlessPsId.Value,
+						}, cancellationToken);
+					else
+						await ic.PermissionSets.Delete(new InstancePermissionSetRequest
+						{
+							PermissionSetId = permlessPsId.Value
+						}, cancellationToken);
+				}));
+		}
+
+		public void CompleteNow() => finishTcs.TrySetResult();
+
+		public Task AbortingConnection(ConnectionAbortReason reason, CancellationToken cancellationToken)
+		{
+			try
+			{
+				Assert.AreEqual(ConnectionAbortReason.ServerRestart, reason);
+				var remaining = Interlocked.Decrement(ref expectedReboots);
+				Assert.IsTrue(remaining >= 0);
+			}
+			catch (Exception ex)
+			{
+				finishTcs.TrySetException(ex);
+			}
+			return Task.CompletedTask;
+		}
+	}
+}

tests/Tgstation.Server.Tests/Live/RateLimitRetryingApiClient.cs

@@ -13,8 +13,18 @@ namespace Tgstation.Server.Tests.Live
 {
 	sealed class RateLimitRetryingApiClient : ApiClient
 	{
-		public RateLimitRetryingApiClient(IHttpClient httpClient, Uri url, ApiHeaders apiHeaders, ApiHeaders tokenRefreshHeaders, bool authless)
-			: base(httpClient, url, apiHeaders, tokenRefreshHeaders, authless)
+		public RateLimitRetryingApiClient(
+			IHttpClient httpClient,
+			Uri url,
+			ApiHeaders apiHeaders,
+			ApiHeaders tokenRefreshHeaders,
+			bool authless)
+			: base(
+				  httpClient,
+				  url,
+				  apiHeaders,
+				  tokenRefreshHeaders,
+				  authless)
 		{
 		}
 

tests/Tgstation.Server.Tests/Live/RateLimitRetryingApiClientFactory.cs

@@ -1,6 +1,7 @@
 using System;
 
 using Tgstation.Server.Api;
+using Tgstation.Server.Api.Models.Response;
 using Tgstation.Server.Client;
 using Tgstation.Server.Common.Http;
 
@@ -8,7 +9,11 @@ namespace Tgstation.Server.Tests.Live
 {
 	sealed class RateLimitRetryingApiClientFactory : IApiClientFactory
 	{
-		public IApiClient CreateApiClient(Uri url, ApiHeaders apiHeaders, ApiHeaders tokenRefreshHeaders, bool authless)
+		public IApiClient CreateApiClient(
+			Uri url,
+			ApiHeaders apiHeaders,
+			ApiHeaders tokenRefreshHeaders,
+			bool authless)
 			=> new RateLimitRetryingApiClient(
 				new HttpClient(),
 				url,

tests/Tgstation.Server.Tests/Live/RawRequestTests.cs

@@ -5,15 +5,25 @@ using System.Net;
 using System.Net.Http;
 using System.Net.Http.Headers;
 using System.Net.Mime;
+using System.Reflection;
 using System.Runtime.InteropServices;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 
+using Microsoft.AspNetCore.Http.Connections;
+using Microsoft.AspNetCore.SignalR.Client;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.DependencyInjection.Extensions;
+using Microsoft.Extensions.Logging;
+
 using Tgstation.Server.Api;
+using Tgstation.Server.Api.Hubs;
 using Tgstation.Server.Api.Models;
+using Tgstation.Server.Api.Models.Request;
 using Tgstation.Server.Api.Models.Response;
 using Tgstation.Server.Client;
+using Tgstation.Server.Client.Extensions;
 using Tgstation.Server.Common.Extensions;
 using Tgstation.Server.Host;
 
@@ -346,12 +356,132 @@ namespace Tgstation.Server.Tests.Live
 			}
 		}
 
+		class FuncProxiedJobsHub : IJobsHub
+		{
+			public Func<JobResponse, CancellationToken, Task> ProxyFunc { get; set; }
+			public Func<ConnectionAbortReason, Task> ErrorFunc { get; set; }
+
+			public Task AbortingConnection(ConnectionAbortReason reason, CancellationToken cancellationToken)
+				=> ErrorFunc(reason);
+
+			public Task ReceiveJobUpdate(JobResponse job, CancellationToken cancellationToken)
+				=> ProxyFunc(job, cancellationToken);
+		}
+
+		static async Task TestSignalRUsage(IServerClientFactory serverClientFactory, IServerClient serverClient, CancellationToken cancellationToken)
+		{
+			// test regular creation works without error
+			var hubConnectionBuilder = new HubConnectionBuilder();
+
+			var tokenRetrivalFunc = () => Task.FromResult("FakeToken");
+
+			hubConnectionBuilder.WithUrl(
+				new Uri(serverClient.Url, Routes.JobsHub),
+				HttpTransportType.ServerSentEvents,
+				options =>
+				{
+					options.AccessTokenProvider = () => tokenRetrivalFunc();
+					((IApiClient)typeof(ServerClient)
+						.GetField(
+							"apiClient",
+							BindingFlags.NonPublic | BindingFlags.Instance)
+						.GetValue(serverClient))
+						.Headers
+						.SetHubConnectionHeaders(options.Headers);
+				});
+
+			hubConnectionBuilder.ConfigureLogging(
+				loggingBuilder =>
+				{
+					loggingBuilder.SetMinimumLevel(LogLevel.Trace);
+					loggingBuilder.AddConsole();
+					loggingBuilder
+						.Services
+						.TryAddEnumerable(
+							ServiceDescriptor.Singleton<ILoggerProvider, HardFailLoggerProvider>());
+				});
+
+			var proxy = new FuncProxiedJobsHub();
+			var errorTcs = new TaskCompletionSource();
+			proxy.ErrorFunc = reason =>
+			{
+				errorTcs.SetException(new Exception($"Aborted: {reason}"));
+				return Task.CompletedTask;
+			};
+
+			HubConnection hubConnection;
+			HardFailLoggerProvider.BlockFails = true;
+			try
+			{
+				await using (hubConnection = hubConnectionBuilder.Build())
+				{
+					Assert.AreEqual(HubConnectionState.Disconnected, hubConnection.State);
+					hubConnection.ProxyOn<IJobsHub>(proxy);
+
+					var exception = await Assert.ThrowsExceptionAsync<HttpRequestException>(() => hubConnection.StartAsync(cancellationToken));
+
+					Assert.AreEqual(HttpStatusCode.Unauthorized, exception.StatusCode);
+					Assert.AreEqual(HubConnectionState.Disconnected, hubConnection.State);
+
+					tokenRetrivalFunc = () => Task.FromResult(serverClient.Token.Bearer);
+					await hubConnection.StartAsync(cancellationToken);
+
+					Assert.AreEqual(HubConnectionState.Connected, hubConnection.State);
+				}
+
+				Assert.AreEqual(HubConnectionState.Disconnected, hubConnection.State);
+
+				Assert.IsFalse(errorTcs.Task.IsCompleted);
+
+				var createRequest = new UserCreateRequest
+				{
+					Enabled = true,
+					Name = "SignalRTestUser",
+					Password = "asdfasdfasdfasdfasdf"
+				};
+
+				var testUser = await serverClient.Users.Create(createRequest, cancellationToken);
+				await using (var testUserClient = await serverClientFactory.CreateFromLogin(serverClient.Url, createRequest.Name, createRequest.Password, cancellationToken: cancellationToken))
+				{
+					errorTcs = new TaskCompletionSource();
+					await using var testUserConn1 = await testUserClient.SubscribeToJobUpdates(proxy, cancellationToken: cancellationToken);
+
+					Assert.IsFalse(errorTcs.Task.IsCompleted);
+
+					await serverClient.Users.Update(new UserUpdateRequest
+					{
+						Id = testUser.Id,
+						Enabled = false,
+					}, cancellationToken);
+
+					// need a second here
+					for (var i = 0; i < 10 && !errorTcs.Task.IsCompleted; ++i)
+						await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken);
+
+					Assert.IsTrue(errorTcs.Task.IsCompleted);
+
+					errorTcs = new TaskCompletionSource();
+					await using var testUserConn2 = await testUserClient.SubscribeToJobUpdates(proxy, cancellationToken: cancellationToken);
+					for (var i = 0; i < 10 && !errorTcs.Task.IsCompleted; ++i)
+						await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken);
+				}
+
+				Assert.IsTrue(errorTcs.Task.IsCompleted);
+				await Assert.ThrowsExceptionAsync<Exception>(() => errorTcs.Task);
+			}
+			finally
+			{
+				HardFailLoggerProvider.BlockFails = false;
+			}
+		}
+
 		public static Task Run(IServerClientFactory clientFactory, IServerClient serverClient, CancellationToken cancellationToken)
 			=> Task.WhenAll(
 				TestRequestValidation(serverClient, cancellationToken),
 				TestOAuthFails(serverClient, cancellationToken),
 				TestServerInformation(clientFactory, serverClient, cancellationToken),
 				TestInvalidTransfers(serverClient, cancellationToken),
-				RegressionTestForLeakedPasswordHashesBug(serverClient, cancellationToken));
+				RegressionTestForLeakedPasswordHashesBug(serverClient, cancellationToken),
+				TestSignalRUsage(clientFactory, serverClient, cancellationToken));
 	}
 }

tests/Tgstation.Server.Tests/Live/TestLiveServer.cs

@@ -255,7 +255,7 @@ namespace Tgstation.Server.Tests.Live
 					return await action();
 				}
 
-				using (var adminClient = await CreateAdminClient(server.Url, cancellationToken))
+				await using (var adminClient = await CreateAdminClient(server.Url, cancellationToken))
 				{
 					// Disabled OAuth test
 					using (var httpClient = new HttpClient())
@@ -336,7 +336,7 @@ namespace Tgstation.Server.Tests.Live
 				await new Host.IO.DefaultIOManager().DeleteDirectory(server.UpdatePath, cancellationToken);
 				serverTask = server.Run(cancellationToken).AsTask();
 
-				using (var adminClient = await CreateAdminClient(server.Url, cancellationToken))
+				await using (var adminClient = await CreateAdminClient(server.Url, cancellationToken))
 				{
 					// test we can't do this without the correct permission
 
@@ -392,7 +392,7 @@ namespace Tgstation.Server.Tests.Live
 			try
 			{
 				var testUpdateVersion = new Version(5, 11, 20);
-				using var adminClient = await CreateAdminClient(server.Url, cancellationToken);
+				await using var adminClient = await CreateAdminClient(server.Url, cancellationToken);
 				await ApiAssert.ThrowsException<ConflictException, ServerUpdateResponse>(
 					() => adminClient.Administration.Update(
 						new ServerUpdateRequest
@@ -442,7 +442,7 @@ namespace Tgstation.Server.Tests.Live
 
 				try
 				{
-					using var controllerClient = await CreateAdminClient(controller.Url, cancellationToken);
+					await using var controllerClient = await CreateAdminClient(controller.Url, cancellationToken);
 
 					var controllerInfo = await controllerClient.ServerInformation(cancellationToken);
 
@@ -544,9 +544,9 @@ namespace Tgstation.Server.Tests.Live
 
 				try
 				{
-					using var controllerClient = await CreateAdminClient(controller.Url, cancellationToken);
-					using var node1Client = await CreateAdminClient(node1.Url, cancellationToken);
-					using var node2Client = await CreateAdminClient(node2.Url, cancellationToken);
+					await using var controllerClient = await CreateAdminClient(controller.Url, cancellationToken);
+					await using var node1Client = await CreateAdminClient(node1.Url, cancellationToken);
+					await using var node2Client = await CreateAdminClient(node2.Url, cancellationToken);
 
 					var controllerInfo = await controllerClient.ServerInformation(cancellationToken);
 
@@ -610,12 +610,12 @@ namespace Tgstation.Server.Tests.Live
 					Assert.AreEqual(newUser.Name, node1User.Name);
 					Assert.AreEqual(newUser.Enabled, node1User.Enabled);
 
-					using var controllerUserClient = await clientFactory.CreateFromLogin(
+					await using var controllerUserClient = await clientFactory.CreateFromLogin(
 						controllerAddress,
 						newUser.Name,
 						"asdfasdfasdfasdf");
 
-					using var node1BadClient = clientFactory.CreateFromToken(node1.Url, controllerUserClient.Token);
+					await using var node1BadClient = clientFactory.CreateFromToken(node1.Url, controllerUserClient.Token);
 					await ApiAssert.ThrowsException<UnauthorizedException, AdministrationResponse>(() => node1BadClient.Administration.Read(cancellationToken));
 
 					// check instance info is not shared
@@ -685,8 +685,8 @@ namespace Tgstation.Server.Tests.Live
 						controller.Run(cancellationToken).AsTask(),
 						node1.Run(cancellationToken).AsTask());
 
-					using var controllerClient2 = await CreateAdminClient(controller.Url, cancellationToken);
-					using var node1Client2 = await CreateAdminClient(node1.Url, cancellationToken);
+					await using var controllerClient2 = await CreateAdminClient(controller.Url, cancellationToken);
+					await using var node1Client2 = await CreateAdminClient(node1.Url, cancellationToken);
 
 					await ApiAssert.ThrowsException<ApiConflictException, ServerUpdateResponse>(() => controllerClient2.Administration.Update(
 						new ServerUpdateRequest
@@ -701,7 +701,7 @@ namespace Tgstation.Server.Tests.Live
 						serverTask,
 						node2.Run(cancellationToken).AsTask());
 
-					using var node2Client2 = await CreateAdminClient(node2.Url, cancellationToken);
+					await using var node2Client2 = await CreateAdminClient(node2.Url, cancellationToken);
 
 					async Task WaitForSwarmServerUpdate2()
 					{
@@ -815,9 +815,9 @@ namespace Tgstation.Server.Tests.Live
 
 				try
 				{
-					using var controllerClient = await CreateAdminClient(controller.Url, cancellationToken);
-					using var node1Client = await CreateAdminClient(node1.Url, cancellationToken);
-					using var node2Client = await CreateAdminClient(node2.Url, cancellationToken);
+					await using var controllerClient = await CreateAdminClient(controller.Url, cancellationToken);
+					await using var node1Client = await CreateAdminClient(node1.Url, cancellationToken);
+					await using var node2Client = await CreateAdminClient(node2.Url, cancellationToken);
 
 					var controllerInfo = await controllerClient.ServerInformation(cancellationToken);
 
@@ -897,7 +897,7 @@ namespace Tgstation.Server.Tests.Live
 					Assert.IsTrue(controllerTask.IsCompleted);
 
 					controllerTask = controller.Run(cancellationToken).AsTask();
-					using var controllerClient2 = await CreateAdminClient(controller.Url, cancellationToken);
+					await using var controllerClient2 = await CreateAdminClient(controller.Url, cancellationToken);
 
 					// node 2 should reconnect once it's health check triggers
 					await Task.WhenAny(
@@ -934,7 +934,7 @@ namespace Tgstation.Server.Tests.Live
 						ErrorCode.SwarmIntegrityCheckFailed);
 
 					node2Task = node2.Run(cancellationToken).AsTask();
-					using var node2Client2 = await CreateAdminClient(node2.Url, cancellationToken);
+					await using var node2Client2 = await CreateAdminClient(node2.Url, cancellationToken);
 
 					// should re-register
 					await Task.WhenAny(
@@ -991,7 +991,7 @@ namespace Tgstation.Server.Tests.Live
 			var serverTask = server.Run(cancellationToken);
 			try
 			{
-				using var adminClient = await CreateAdminClient(server.Url, cancellationToken);
+				await using var adminClient = await CreateAdminClient(server.Url, cancellationToken);
 
 				var instanceManagerTest = new InstanceManagerTest(adminClient, server.Directory);
 				var instance = await instanceManagerTest.CreateTestInstance("TgTestInstance", cancellationToken);
@@ -1273,7 +1273,29 @@ namespace Tgstation.Server.Tests.Live
 			{
 				Api.Models.Instance instance;
 				long initialStaged, initialActive;
-				using (var adminClient = await CreateAdminClient(server.Url, cancellationToken))
+				await using var firstAdminClient = await CreateAdminClient(server.Url, cancellationToken);
+
+				async ValueTask<IServerClient> CreateUserWithNoInstancePerms()
+				{
+					var createRequest = new UserCreateRequest()
+					{
+						Name = "SomePermlessChum",
+						Password = "alidfjuwh84322r4yrkajhfdqh38hrfiouw4",
+						Enabled = true,
+						PermissionSet = new PermissionSet
+						{
+							InstanceManagerRights = InstanceManagerRights.Read,
+						}
+					};
+
+					var user = await firstAdminClient.Users.Create(createRequest, cancellationToken);
+					Assert.IsTrue(user.Enabled);
+
+					return await clientFactory.CreateFromLogin(server.Url, createRequest.Name, createRequest.Password, cancellationToken: cancellationToken);
+				}
+
+				var jobsHubTest = new JobsHubTests(firstAdminClient, await CreateUserWithNoInstancePerms());
+				Task jobsHubTestTask;
 				{
 					if (server.DumpOpenApiSpecpath)
 					{
@@ -1302,21 +1324,23 @@ namespace Tgstation.Server.Tests.Live
 						}
 					}
 
-					var rootTest = FailFast(RawRequestTests.Run(clientFactory, adminClient, cancellationToken));
-					var adminTest = FailFast(new AdministrationTest(adminClient.Administration).Run(cancellationToken));
-					var usersTest = FailFast(new UsersTest(adminClient).Run(cancellationToken));
-					var instanceManagerTest = new InstanceManagerTest(adminClient, server.Directory);
+					var rootTest = FailFast(RawRequestTests.Run(clientFactory, firstAdminClient, cancellationToken));
+					var adminTest = FailFast(new AdministrationTest(firstAdminClient.Administration).Run(cancellationToken));
+					var usersTest = FailFast(new UsersTest(firstAdminClient).Run(cancellationToken));
+
+					jobsHubTestTask = FailFast(jobsHubTest.Run(cancellationToken));
+					var instanceManagerTest = new InstanceManagerTest(firstAdminClient, server.Directory);
 					var compatInstanceTask = instanceManagerTest.CreateTestInstance("CompatTestsInstance", cancellationToken);
 					instance = await instanceManagerTest.CreateTestInstance("LiveTestsInstance", cancellationToken);
 					var compatInstance = await compatInstanceTask;
 					var instancesTest = FailFast(instanceManagerTest.RunPreTest(cancellationToken));
 					Assert.IsTrue(Directory.Exists(instance.Path));
-					var instanceClient = adminClient.Instances.CreateClient(instance);
+					var instanceClient = firstAdminClient.Instances.CreateClient(instance);
 
 					Assert.IsTrue(Directory.Exists(instanceClient.Metadata.Path));
 
 					var instanceTest = new InstanceTest(
-							adminClient.Instances,
+							firstAdminClient.Instances,
 							fileDownloader,
 							GetInstanceManager(),
 							(ushort)server.Url.Port);
@@ -1329,7 +1353,7 @@ namespace Tgstation.Server.Tests.Live
 									new PlatformIdentifier().IsWindows
 										? new Version(510, 1346)
 										: new Version(512, 1451), // http://www.byond.com/forum/?forum=5&command=search&scope=local&text=resolved%3a512.1451
-									adminClient.Instances.CreateClient(compatInstance),
+									firstAdminClient.Instances.CreateClient(compatInstance),
 									compatDMPort,
 									compatDDPort,
 									server.HighPriorityDreamDaemon,
@@ -1364,7 +1388,8 @@ namespace Tgstation.Server.Tests.Live
 					initialActive = dd.ActiveCompileJob.Id.Value;
 					initialStaged = dd.StagedCompileJob.Id.Value;
 
-					await adminClient.Administration.Restart(cancellationToken);
+					jobsHubTest.ExpectShutdown();
+					await firstAdminClient.Administration.Restart(cancellationToken);
 				}
 
 				await Task.WhenAny(serverTask, Task.Delay(TimeSpan.FromMinutes(1), cancellationToken));
@@ -1412,8 +1437,9 @@ namespace Tgstation.Server.Tests.Live
 
 				// chat bot start and DD reattach test
 				serverTask = server.Run(cancellationToken).AsTask();
-				using (var adminClient = await CreateAdminClient(server.Url, cancellationToken))
+				await using (var adminClient = await CreateAdminClient(server.Url, cancellationToken))
 				{
+					await jobsHubTest.WaitForReconnect(cancellationToken);
 					var instanceClient = adminClient.Instances.CreateClient(instance);
 
 					var jobs = await instanceClient.Jobs.ListActive(null, cancellationToken);
@@ -1478,6 +1504,7 @@ namespace Tgstation.Server.Tests.Live
 
 					Assert.AreEqual(WatchdogStatus.Offline, dd.Status);
 
+					jobsHubTest.ExpectShutdown();
 					await adminClient.Administration.Restart(cancellationToken);
 				}
 
@@ -1496,7 +1523,6 @@ namespace Tgstation.Server.Tests.Live
 							.Select(e => instanceClient.Jobs.GetId(e, cancellationToken))
 							.ToList();
 
-						
 						jobs = (await ValueTaskExtensions.WhenAll(getTasks))
 							.Where(x => x.StartedAt.Value > preStartupTime)
 						.ToList();
@@ -1515,10 +1541,11 @@ namespace Tgstation.Server.Tests.Live
 				serverTask = server.Run(cancellationToken).AsTask();
 				long expectedCompileJobId, expectedStaged;
 				var edgeByond = await ByondTest.GetEdgeVersion(fileDownloader, cancellationToken);
-				using (var adminClient = await CreateAdminClient(server.Url, cancellationToken))
+				await using (var adminClient = await CreateAdminClient(server.Url, cancellationToken))
 				{
 					var instanceClient = adminClient.Instances.CreateClient(instance);
 					await WaitForInitialJobs(instanceClient);
+					await jobsHubTest.WaitForReconnect(cancellationToken);
 
 					var dd = await instanceClient.DreamDaemon.Read(cancellationToken);
 
@@ -1554,6 +1581,7 @@ namespace Tgstation.Server.Tests.Live
 					await wdt.WaitForJob(compileJob, 30, false, null, cancellationToken);
 					expectedStaged = compileJob.Id.Value;
 
+					jobsHubTest.ExpectShutdown();
 					await adminClient.Administration.Restart(cancellationToken);
 				}
 
@@ -1562,10 +1590,11 @@ namespace Tgstation.Server.Tests.Live
 
 				// post/entity deletion tests
 				serverTask = server.Run(cancellationToken).AsTask();
-				using (var adminClient = await CreateAdminClient(server.Url, cancellationToken))
+				await using (var adminClient = await CreateAdminClient(server.Url, cancellationToken))
 				{
 					var instanceClient = adminClient.Instances.CreateClient(instance);
 					await WaitForInitialJobs(instanceClient);
+					await jobsHubTest.WaitForReconnect(cancellationToken);
 
 					var currentDD = await instanceClient.DreamDaemon.Read(cancellationToken);
 					Assert.AreEqual(expectedCompileJobId, currentDD.ActiveCompileJob.Id.Value);
@@ -1581,6 +1610,9 @@ namespace Tgstation.Server.Tests.Live
 					await new ChatTest(instanceClient.ChatBots, adminClient.Instances, instanceClient.Jobs, instance).RunPostTest(cancellationToken);
 					await repoTest;
 
+					jobsHubTest.CompleteNow();
+					await jobsHubTestTask;
+
 					await new InstanceManagerTest(adminClient, server.Directory).RunPostTest(instance, cancellationToken);
 				}
 			}

tests/Tgstation.Server.Tests/Tgstation.Server.Tests.csproj

@@ -5,6 +5,10 @@
     <TargetFramework>$(TgsFrameworkVersion)</TargetFramework>
   </PropertyGroup>
 
+  <ItemGroup>
+    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="7.0.0" />
+  </ItemGroup>
+
   <ItemGroup>
     <ProjectReference Include="..\..\src\Tgstation.Server.Client\Tgstation.Server.Client.csproj" />
     <ProjectReference Include="..\..\src\Tgstation.Server.Host.Watchdog\Tgstation.Server.Host.Watchdog.csproj" />

tools/Tgstation.Server.Migrator/Tgstation.Server.Migrator.csproj

@@ -1,4 +1,4 @@
-<Project Sdk="Microsoft.NET.Sdk">
+<Project Sdk="Microsoft.NET.Sdk">
   <Import Project="../../build/Common.props" />
 
   <PropertyGroup>
@@ -13,6 +13,7 @@
 
   <ItemGroup>
     <PackageReference Include="Core.System.ServiceProcess" Version="2.0.1" />
+    <PackageReference Include="Microsoft.Bcl.AsyncInterfaces" Version="7.0.0" />
   </ItemGroup>
 
   <ItemGroup>