Repository: hikari-py/hikari
Pull request #1204 (2022-07-06 01:08:33)

Improve components lifetime
# Short summary

Bit of a rewrite of hikari internals to improve the lifetime, stability and consistency of the components.

# Extended summary

## Breaking changes
- `GatewayBot.join`'s `until_close` argument removed.
- `GatewayShardImpl.get_user_id` is no longer async and will now always be available.

## Bugfixes
- `GatewayShardImpl` can now be instantiated out of an async environment for consistency with other components.
- Correct signal handling in `RESTBot`.


## Improvements
- `GatewayBot`:
  - General speedups.
  - Fix a lot of edge cases of hard crashes if the application shuts unexpectedly.
  - More consistent signal handling.
  - `run`' `shard_ids` argument can now be a `typing.Sequence`.
  - Improved logging.
- `RESTBot`:
  - Consistent signal handling inline with `GatewayBot`.
  - Improved logging.
  - Improved loop closing.
- `GatewayShardImpl`:
  - New `is_connected` property to determine whether the shard is connected to the gateway.
  - Faster websocket pulling and heartbeating.
  - Improved error handling.
  - New gateway reconnect logic to account for `resume_gateway_url`.
  - Rate limiting changes:
    - Chunking no longer has its own special ratelimit. Now it is shared with the rest of
    "non-priority" packages sent, which is of 117/60s (3 less than the hard limit).
      - "priority" packages currently only include hearbeating.

---

*To @FasterSpeeding: I didnt realize until now how many "small fixes" there are in this pr, which makes it a bit hard to review. If you want, I could make smaller separate PRs for some of the issues if they are better for you to review* 

davfsa: Implemented https://github.com/discord/discord-api-docs/pull/5282 as part of https://github.com/hikari-py/hikari/pull/1204/commits/5c1b5e7a1a6e847ca9693b4d53711d870e9b4d41.

If this PR were not to be merged before the 7th of August, I will open another smaller PR to merge directly into master (or if @FasterSpeeding rathers)

And bump for review :)
davfsa: @FasterSpeeding Bump :)

Patch:
diff --git a/changes/1204.breaking.md b/changes/1204.breaking.md
new file mode 100644
index 00000000..3b4a2eea
--- /dev/null
+++ b/changes/1204.breaking.md
@@ -0,0 +1,3 @@
+Lifetime improvements breaking changes:
+- `GatewayBot.join`'s `until_close` argument removed.
+- `GatewayShardImpl.get_user_id` is no longer async and will now always be available.
diff --git a/changes/1204.bugfix.md b/changes/1204.bugfix.md
new file mode 100644
index 00000000..f483e19c
--- /dev/null
+++ b/changes/1204.bugfix.md
@@ -0,0 +1,3 @@
+Lifetime improvements bugfixes:
+- `GatewayShardImpl` can now be instantiated out of an async environment for consistency with other components.
+- Correct signal handling in `RESTBot`.
diff --git a/changes/1204.feature.md b/changes/1204.feature.md
new file mode 100644
index 00000000..a2f475ca
--- /dev/null
+++ b/changes/1204.feature.md
@@ -0,0 +1,20 @@
+Improve components lifetimes:
+- `GatewayBot`:
+  - General speedups.
+  - Fix a lot of edge cases of hard crashes if the application shuts unexpectedly.
+  - More consistent signal handling.
+  - `run`' `shard_ids` argument can now be a `typing.Sequence`.
+  - Improved logging.
+- `RESTBot`:
+  - Consistent signal handling inline with `GatewayBot`.
+  - Improved logging.
+  - Improved loop closing.
+- `GatewayShardImpl`:
+  - New `is_connected` property to determine whether the shard is connected to the gateway.
+  - Faster websocket pulling and heartbeating.
+  - Improved error handling.
+  - New gateway reconnect logic to account for `resume_gateway_url`.
+  - Rate limiting changes:
+    - Chunking no longer has its own special ratelimit. Now it is shared with the rest of
+    "non-priority" packages sent, which is of 117/60s (3 less than the hard limit).
+      - "priority" packages currently only include heartbeating.
diff --git a/hikari/api/event_manager.py b/hikari/api/event_manager.py
index 8e2d3464..05ac7ea4 100644
--- a/hikari/api/event_manager.py
+++ b/hikari/api/event_manager.py
@@ -376,7 +376,7 @@ def get_listeners(
 
         Returns
         -------
-        typing.Collection[typing.Callable[[T], typing.Coroutine[typing.Any, typing.Any, builtins.None]]
+        typing.Collection[typing.Callable[[T], typing.Coroutine[typing.Any, typing.Any, builtins.None]]]
             A copy of the collection of listeners for the event. Will return
             an empty collection if nothing is registered.
 
diff --git a/hikari/api/shard.py b/hikari/api/shard.py
index 4f5fcffe..b9899b1a 100644
--- a/hikari/api/shard.py
+++ b/hikari/api/shard.py
@@ -107,13 +107,12 @@ def intents(self) -> intents_.Intents:
     @property
     @abc.abstractmethod
     def is_alive(self) -> bool:
-        """Return `builtins.True` if the shard is alive and connected.
+        """Whether the shard is alive."""
 
-        Returns
-        -------
-        builtins.bool
-            `builtins.True` if connected, or `builtins.False` if not.
-        """
+    @property
+    @abc.abstractmethod
+    def is_connected(self) -> bool:
+        """Whether the shard is connected."""
 
     @property
     @abc.abstractmethod
@@ -127,12 +126,9 @@ def shard_count(self) -> int:
         """
 
     @abc.abstractmethod
-    async def get_user_id(self) -> snowflakes.Snowflake:
+    def get_user_id(self) -> snowflakes.Snowflake:
         """Return the user ID.
 
-        If the shard has not connected fully yet, this should wait until the ID
-        is set before returning.
-
         Returns
         -------
         hikari.snowflakes.Snowflake
diff --git a/hikari/impl/bot.py b/hikari/impl/bot.py
index 6107facd..073f591c 100644
--- a/hikari/impl/bot.py
+++ b/hikari/impl/bot.py
@@ -30,10 +30,7 @@
 import datetime
 import logging
 import math
-import signal
 import sys
-import threading
-import traceback
 import types
 import typing
 import warnings
@@ -54,6 +51,7 @@
 from hikari.impl import shard as shard_impl
 from hikari.impl import voice as voice_impl
 from hikari.internal import aio
+from hikari.internal import signals
 from hikari.internal import time
 from hikari.internal import ux
 
@@ -75,58 +73,6 @@
 _LOGGER: typing.Final[logging.Logger] = logging.getLogger("hikari.bot")
 
 
-async def _gather(coros: typing.Iterator[typing.Awaitable[typing.Any]]) -> None:
-    # Calling asyncio.gather outside of a running event loop isn't safe and
-    # will lead to RuntimeErrors in later versions of python, so this call is
-    # kept within a coroutine function.
-    await asyncio.gather(*coros)
-
-
-def _destroy_loop(loop: asyncio.AbstractEventLoop) -> None:
-    async def murder(future: asyncio.Future[typing.Any]) -> None:
-        # These include _GatheringFuture which must be awaited if the children
-        # throw an asyncio.CancelledError, otherwise it will spam logs with warnings
-        # about exceptions not being retrieved before GC.
-        try:
-            _LOGGER.log(ux.TRACE, "killing %s", future)
-            future.cancel()
-            await future
-        except asyncio.CancelledError:
-            pass
-        except Exception as ex:
-            loop.call_exception_handler(
-                {
-                    "message": "Future raised unexpected exception after requesting cancellation",
-                    "exception": ex,
-                    "future": future,
-                }
-            )
-
-    remaining_tasks = [t for t in asyncio.all_tasks(loop) if not t.done()]
-
-    if remaining_tasks:
-        _LOGGER.debug("terminating %s remaining tasks forcefully", len(remaining_tasks))
-        loop.run_until_complete(_gather((murder(task) for task in remaining_tasks)))
-    else:
-        _LOGGER.debug("No remaining tasks exist, good job!")
-
-    if sys.version_info >= (3, 9):
-        _LOGGER.debug("shutting down default executor")
-        try:
-            # This seems to raise a NotImplementedError when running with uvloop.
-            loop.run_until_complete(loop.shutdown_default_executor())
-        except NotImplementedError:
-            pass
-
-    _LOGGER.debug("shutting down asyncgens")
-    loop.run_until_complete(loop.shutdown_asyncgens())
-
-    _LOGGER.debug("closing event loop")
-    loop.close()
-    # Closed loops cannot be re-used so it should also be un-set.
-    asyncio.set_event_loop(None)
-
-
 def _validate_activity(activity: undefined.UndefinedNoneOr[presences.Activity]) -> None:
     # This seems to cause confusion for a lot of people, so lets add some warnings into the mix.
 
@@ -313,7 +259,6 @@ class GatewayBot(traits.GatewayBotAware):
         "_executor",
         "_http_settings",
         "_intents",
-        "_is_alive",
         "_proxy_settings",
         "_rest",
         "_shards",
@@ -345,9 +290,8 @@ def __init__(
         self.print_banner(banner, allow_color, force_color)
 
         # Settings and state
-        self._closing_event: typing.Optional[asyncio.Event] = None
         self._closed_event: typing.Optional[asyncio.Event] = None
-        self._is_alive = False
+        self._closing_event: typing.Optional[asyncio.Event] = None
         self._executor = executor
         self._http_settings = http_settings if http_settings is not None else config_impl.HTTPSettings()
         self._intents = intents
@@ -450,32 +394,28 @@ def rest(self) -> rest_.RESTClient:
 
     @property
     def is_alive(self) -> bool:
-        return self._is_alive
+        return self._closed_event is not None
 
     def _check_if_alive(self) -> None:
-        if not self._is_alive:
+        if not self._closed_event:
             raise errors.ComponentStateConflictError("bot is not running so it cannot be interacted with")
 
     def get_me(self) -> typing.Optional[users_.OwnUser]:
         return self._cache.get_me()
 
     async def close(self) -> None:
-        self._check_if_alive()
-        await self._close()
-
-    async def _close(self) -> None:
-        if self._closed_event:  # Closing is in progress from another call, wait for that to complete.
-            await self._closed_event.wait()
-            return
+        if not self._closed_event or not self._closing_event:
+            raise errors.ComponentStateConflictError("Cannot close an inactive bot")
 
-        if self._closing_event is None:  # If closing event is None then this is already closed.
+        if self._closing_event.is_set():
+            await self.join()
             return
 
-        _LOGGER.debug("bot requested to shutdown")
-        self._closed_event = asyncio.Event()
+        _LOGGER.info("bot requested to shut down")
         self._closing_event.set()
-        self._closing_event = None
-        dispatch_events = self._is_alive
+
+        await self._event_manager.dispatch(self._event_factory.deserialize_stopping_event())
+        _LOGGER.log(ux.TRACE, "StoppingEvent dispatch completed, now beginning termination")
 
         loop = asyncio.get_running_loop()
 
@@ -487,36 +427,31 @@ async def handle(name: str, awaitable: typing.Awaitable[typing.Any]) -> None:
             except Exception as ex:
                 loop.call_exception_handler(
                     {
-                        "message": f"{name} raised an exception during shutdown",
+                        "message": f"{name} raised an exception during shut down",
                         "future": future,
                         "exception": ex,
                     }
                 )
 
-        if dispatch_events:
-            await self._event_manager.dispatch(self._event_factory.deserialize_stopping_event())
+        shards = tuple((handle(f"shard {s.id}", s.close()) for s in self._shards.values() if s.is_alive))
 
-        _LOGGER.log(ux.TRACE, "StoppingEvent dispatch completed, now beginning termination")
-
-        calls = [
-            ("rest", self._rest.close()),
-            ("voice handler", self._voice.close()),
-            *((f"shard {s.id}", s.close()) for s in self._shards.values()),
-        ]
-
-        for coro in asyncio.as_completed([handle(*pair) for pair in calls]):
+        for coro in asyncio.as_completed(shards):
             await coro
 
+        await handle("voice handler", self._voice.close())
+        await handle("rest", self._rest.close())
+
         # Clear out cache and shard map
         self._cache.clear()
         self._shards.clear()
-        self._is_alive = False
 
-        if dispatch_events:
-            await self._event_manager.dispatch(self._event_factory.deserialize_stopped_event())
+        await self._event_manager.dispatch(self._event_factory.deserialize_stopped_event())
 
         self._closed_event.set()
         self._closed_event = None
+        self._closing_event = None
+
+        _LOGGER.info("bot shut down successfully")
 
     def dispatch(self, event: base_events.Event) -> asyncio.Future[typing.Any]:
         """Dispatch an event.
@@ -619,7 +554,7 @@ def get_listeners(
 
         Returns
         -------
-        typing.Collection[typing.Callable[[EventT], typing.Coroutine[typing.Any, typing.Any, builtins.None]]
+        typing.Collection[typing.Callable[[EventT], typing.Coroutine[typing.Any, typing.Any, builtins.None]]]
             A copy of the collection of listeners for the event. Will return
             an empty collection if nothing is registered.
 
@@ -627,14 +562,11 @@ def get_listeners(
         """
         return self._event_manager.get_listeners(event_type, polymorphic=polymorphic)
 
-    async def join(self, until_close: bool = True) -> None:
-        self._check_if_alive()
-
-        awaitables: typing.List[typing.Awaitable[typing.Any]] = [s.join() for s in self._shards.values()]
-        if until_close and self._closing_event:  # If closing event is None then this is already closing.
-            awaitables.append(self._closing_event.wait())
+    async def join(self) -> None:
+        if not self._closed_event:
+            raise errors.ComponentStateConflictError("Cannot wait for an inactive bot to join")
 
-        await aio.first_completed(*awaitables)
+        await aio.first_completed(self._closed_event.wait(), *(s.join() for s in self._shards.values()))
 
     def listen(
         self,
@@ -728,7 +660,7 @@ def run(
         large_threshold: int = 250,
         propagate_interrupts: bool = False,
         status: presences.Status = presences.Status.ONLINE,
-        shard_ids: typing.Optional[typing.AbstractSet[int]] = None,
+        shard_ids: typing.Optional[typing.Sequence[int]] = None,
         shard_count: typing.Optional[int] = None,
     ) -> None:
         """Start the bot, wait for all shards to become ready, and then return.
@@ -757,7 +689,7 @@ def run(
         close_loop : builtins.bool
             Defaults to `builtins.True`. If `builtins.True`, then once the bot
             enters a state where all components have shut down permanently
-            during application shutdown, then all asyncgens and background tasks
+            during application shut down, then all asyncgens and background tasks
             will be destroyed, and the event loop will be shut down.
 
             This will wait until all `hikari`-owned `aiohttp` connectors have
@@ -786,7 +718,7 @@ def run(
         ignore_session_start_limit : builtins.bool
             Defaults to `builtins.False`. If `builtins.False`, then attempting
             to start more sessions than you are allowed in a 24 hour window
-            will throw a `hikari.errors.GatewayError` rather than going ahead
+            will throw a `RuntimeError` rather than going ahead
             and hitting the IDENTIFY limit, which may result in your token
             being reset. Setting to `builtins.True` disables this behavior.
         large_threshold : builtins.int
@@ -802,11 +734,13 @@ def run(
             determine what kind of interrupt the application received after
             it closes. When `builtins.False`, nothing is raised and the call
             will terminate cleanly and silently where possible instead.
-        shard_ids : typing.Optional[typing.AbstractSet[builtins.int]]
+        shard_ids : typing.Optional[typing.Sequence[builtins.int]]
             The shard IDs to create shards for. If not `builtins.None`, then
             a non-`None` `shard_count` must ALSO be provided. Defaults to
             `builtins.None`, which means the Discord-recommended count is used
             for your application instead.
+
+            Note that the sequence will be de-duplicated.
         shard_count : typing.Optional[builtins.int]
             The number of shards to use in the entire distributed application.
             Defaults to `builtins.None` which results in the count being
@@ -822,14 +756,13 @@ def run(
         builtins.TypeError
             If `shard_ids` is passed without `shard_count`.
         """
-        if self._is_alive:
+        if self._closed_event:
             raise errors.ComponentStateConflictError("bot is already running")
 
         if shard_ids is not None and shard_count is None:
             raise TypeError("'shard_ids' must be passed with 'shard_count'")
 
         loop = aio.get_or_make_loop()
-        signals = ("SIGINT", "SIGTERM")
 
         if asyncio_debug:
             loop.set_debug(True)
@@ -841,100 +774,44 @@ def run(
             except AttributeError:
                 _LOGGER.log(ux.TRACE, "cannot set coroutine tracking depth for sys, no functionality exists for this")
 
-        # Throwing this in the handler will lead to lots of fun OS specific shenanigans. So, lets just
-        # cache it for later, I guess.
-        interrupt: typing.Optional[errors.HikariInterrupt] = None
-        loop_thread_id = threading.get_native_id()
-
-        def handle_os_interrupt(signum: int, frame: typing.Optional[types.FrameType]) -> None:
-            # If we use a POSIX system, then raising an exception in here works perfectly and shuts the loop down
-            # with an exception, which is good.
-            # Windows, however, is special on this front. On Windows, the exception is caught by whatever was
-            # currently running on the event loop at the time, which is annoying for us, as this could be fired into
-            # the task for an event dispatch, for example, which is a guarded call that is never waited for by design.
-
-            # We can't always safely intercept this either, as Windows does not allow us to use asyncio loop
-            # signal listeners (since Windows doesn't have kernel-level signals, only emulated system calls
-            # for a remote few standard C signal types). Thus, the best solution here is to set the close bit
-            # instead, which will let the bot start to clean itself up as if the user closed it manually via a call
-            # to `bot.close()`.
-            nonlocal interrupt
-            signame = signal.strsignal(signum)
-            assert signame is not None  # Will always be True
-
-            interrupt = errors.HikariInterrupt(signum, signame)
-            # The loop may or may not be running, depending on the state of the application when this occurs.
-            # Signals on POSIX only occur on the main thread usually, too, so we need to ensure this is
-            # threadsafe if we want the user's application to still shut down if on a separate thread.
-            # We log native thread IDs purely for debugging purposes.
-            if _LOGGER.isEnabledFor(ux.TRACE):
-                _LOGGER.log(
-                    ux.TRACE,
-                    "interrupt %s occurred on thread %s, bot on thread %s will be notified to shut down shortly\n"
-                    "Stacktrace for developer sanity:\n%s",
-                    signum,
-                    threading.get_native_id(),
-                    loop_thread_id,
-                    "".join(traceback.format_stack(frame)),
+        try:
+            with signals.handle_interrupts(
+                enabled=enable_signal_handlers,
+                loop=loop,
+                propagate_interrupts=propagate_interrupts,
+            ):
+                loop.run_until_complete(
+                    self.start(
+                        activity=activity,
+                        afk=afk,
+                        check_for_updates=check_for_updates,
+                        idle_since=idle_since,
+                        ignore_session_start_limit=ignore_session_start_limit,
+                        large_threshold=large_threshold,
+                        shard_ids=shard_ids,
+                        shard_count=shard_count,
+                        status=status,
+                    )
                 )
 
-            asyncio.run_coroutine_threadsafe(self._set_close_flag(signame, signum), loop)
+                loop.run_until_complete(self.join())
 
-        if enable_signal_handlers is None:
-            # Signal handlers can only be registered on the main thread so we
-            # only default to True if this is the case.
-            enable_signal_handlers = threading.current_thread() is threading.main_thread()
+        finally:
+            if self._closing_event:
+                if self._closing_event.is_set():
+                    loop.run_until_complete(self._closing_event.wait())
+                else:
+                    loop.run_until_complete(self.close())
 
-        if enable_signal_handlers:
-            for sig in signals:
-                try:
-                    signum = getattr(signal, sig)
-                    signal.signal(signum, handle_os_interrupt)
-                except AttributeError:
-                    _LOGGER.log(ux.TRACE, "signal %s is not implemented on your platform", sig)
+            if close_passed_executor and self._executor is not None:
+                _LOGGER.debug("shutting down executor %s", self._executor)
+                self._executor.shutdown(wait=True)
+                self._executor = None
 
-        try:
-            loop.run_until_complete(
-                self.start(
-                    activity=activity,
-                    afk=afk,
-                    check_for_updates=check_for_updates,
-                    idle_since=idle_since,
-                    ignore_session_start_limit=ignore_session_start_limit,
-                    large_threshold=large_threshold,
-                    shard_ids=shard_ids,
-                    shard_count=shard_count,
-                    status=status,
-                )
-            )
-
-            loop.run_until_complete(self.join())
+            if close_loop:
+                aio.destroy_loop(loop, _LOGGER)
 
-        finally:
-            try:
-                loop.run_until_complete(self._close())
-
-                if close_passed_executor and self._executor is not None:
-                    _LOGGER.debug("shutting down executor %s", self._executor)
-                    self._executor.shutdown(wait=True)
-                    self._executor = None
-            finally:
-                if enable_signal_handlers:
-                    for sig in signals:
-                        try:
-                            signum = getattr(signal, sig)
-                            signal.signal(signum, signal.SIG_DFL)
-                        except AttributeError:
-                            # Signal not implemented probably. We should have logged this earlier.
-                            pass
-
-                if close_loop:
-                    _destroy_loop(loop)
-
-                _LOGGER.info("successfully terminated")
-
-                if propagate_interrupts and interrupt is not None:
-                    raise interrupt
+            _LOGGER.info("successfully terminated")
 
     async def start(
         self,
@@ -945,7 +822,7 @@ async def start(
         idle_since: typing.Optional[datetime.datetime] = None,
         ignore_session_start_limit: bool = False,
         large_threshold: int = 250,
-        shard_ids: typing.Optional[typing.AbstractSet[int]] = None,
+        shard_ids: typing.Optional[typing.Sequence[int]] = None,
         shard_count: typing.Optional[int] = None,
         status: presences.Status = presences.Status.ONLINE,
     ) -> None:
@@ -968,18 +845,20 @@ async def start(
         ignore_session_start_limit : builtins.bool
             Defaults to `builtins.False`. If `builtins.False`, then attempting
             to start more sessions than you are allowed in a 24 hour window
-            will throw a `hikari.errors.GatewayError` rather than going ahead
+            will throw a `RuntimeError` rather than going ahead
             and hitting the IDENTIFY limit, which may result in your token
             being reset. Setting to `builtins.True` disables this behavior.
         large_threshold : builtins.int
             Threshold for members in a guild before it is treated as being
             "large" and no longer sending member details in the `GUILD CREATE`
             event. Defaults to `250`.
-        shard_ids : typing.Optional[typing.AbstractSet[builtins.int]]
+        shard_ids : typing.Optional[typing.Sequence[builtins.int]]
             The shard IDs to create shards for. If not `builtins.None`, then
             a non-`None` `shard_count` must ALSO be provided. Defaults to
             `builtins.None`, which means the Discord-recommended count is used
             for your application instead.
+
+            Note that the sequence will be de-duplicated.
         shard_count : typing.Optional[builtins.int]
             The number of shards to use in the entire distributed application.
             Defaults to `builtins.None` which results in the count being
@@ -995,7 +874,7 @@ async def start(
         builtins.TypeError
             If `shard_ids` is passed without `shard_count`.
         """
-        if self._is_alive:
+        if self._closed_event:
             raise errors.ComponentStateConflictError("bot is already running")
 
         if shard_ids is not None and shard_count is None:
@@ -1004,10 +883,8 @@ async def start(
         _validate_activity(activity)
 
         start_time = time.monotonic()
-        self._rest.start()
-        self._voice.start()
+        self._closed_event = asyncio.Event()
         self._closing_event = asyncio.Event()
-        self._is_alive = True
 
         if check_for_updates:
             asyncio.create_task(
@@ -1015,13 +892,18 @@ async def start(
                 name="check for package updates",
             )
 
-        requirements = await self._rest.fetch_gateway_bot_info()
+        self._rest.start()
+        self._voice.start()
+
         await self._event_manager.dispatch(self._event_factory.deserialize_starting_event())
+        requirements = await self._rest.fetch_gateway_bot_info()
 
         if shard_count is None:
             shard_count = requirements.shard_count
         if shard_ids is None:
-            shard_ids = set(range(shard_count))
+            shard_ids = tuple(range(shard_count))
+        else:
+            shard_ids = tuple(dict.fromkeys(shard_ids))
 
         if requirements.session_start_limit.remaining < len(shard_ids) and not ignore_session_start_limit:
             _LOGGER.critical(
@@ -1034,7 +916,7 @@ async def start(
                 "s" if requirements.session_start_limit.remaining != 1 else "",
                 requirements.session_start_limit.reset_at,
             )
-            raise errors.GatewayError("Attempted to start more sessions than were allowed in the given time-window")
+            raise RuntimeError("Attempted to start more sessions than were allowed in the given time-window")
 
         _LOGGER.info(
             "you can start %s session%s before the next window which starts at %s; planning to start %s session%s... ",
@@ -1045,50 +927,29 @@ async def start(
             "s" if len(shard_ids) != 1 else "",
         )
 
-        for window_start in range(0, shard_count, requirements.session_start_limit.max_concurrency):
-            window = [
-                candidate_shard_id
-                for candidate_shard_id in range(
-                    window_start, window_start + requirements.session_start_limit.max_concurrency
-                )
-                if candidate_shard_id in shard_ids
-            ]
+        max_concurrency = requirements.session_start_limit.max_concurrency
+        while shard_ids:
+            window = shard_ids[:max_concurrency]
+            shard_ids = shard_ids[max_concurrency:]
 
-            if not window:
-                continue
             if self._shards:
-                close_waiter = asyncio.create_task(self._closing_event.wait())
-                shard_joiners = [s.join() for s in self._shards.values()]
+                _LOGGER.info("the next startup window is in 5 seconds, please wait...")
 
                 try:
-                    # Attempt to wait for all started shards, for 5 seconds, along with the close
-                    # waiter.
-                    # If the close flag is set (i.e. user invoked bot.close), or one or more shards
-                    # die in this time, we shut down immediately.
-                    # If we time out, the joining tasks get discarded and we spin up the next
-                    # block of shards, if applicable.
-                    _LOGGER.info("the next startup window is in 5 seconds, please wait...")
-                    await aio.first_completed(aio.all_of(*shard_joiners, timeout=5), close_waiter)
-
-                    if not close_waiter.cancelled():
-                        _LOGGER.info("requested to shut down during startup of shards")
-                    else:
-                        _LOGGER.critical("one or more shards shut down unexpectedly during bot startup")
-                    return
+                    await aio.first_completed(
+                        self._closing_event.wait(), *(shard.join() for shard in self._shards.values()), timeout=5
+                    )
 
-                except asyncio.TimeoutError:
-                    # If any shards stopped silently, we should close.
-                    if any(not s.is_alive for s in self._shards.values()):
-                        _LOGGER.warning("one of the shards has been manually shut down (no error), will now shut down")
-                        await self._close()
+                    if self._closing_event.is_set():
                         return
-                    # new window starts.
 
-                except Exception as ex:
-                    _LOGGER.critical("an exception occurred in one of the started shards during bot startup: %r", ex)
-                    raise
+                    _LOGGER.critical("one or more shards closed while starting; shutting down")
+                    raise RuntimeError("One or more shards closed while starting")
+                except asyncio.TimeoutError:
+                    # new window starts.
+                    pass
 
-            await aio.all_of(
+            gather = asyncio.gather(
                 *(
                     self._start_one_shard(
                         activity=activity,
@@ -1096,16 +957,16 @@ async def start(
                         idle_since=idle_since,
                         status=status,
                         large_threshold=large_threshold,
-                        shard_id=candidate_shard_id,
+                        shard_id=shard_id,
                         shard_count=shard_count,
                         url=requirements.url,
-                        closing_event=self._closing_event,
                     )
-                    for candidate_shard_id in window
-                    if candidate_shard_id in shard_ids
+                    for shard_id in window
                 )
             )
 
+            await aio.first_completed(self._closing_event.wait(), gather)
+
         await self._event_manager.dispatch(self._event_factory.deserialize_started_event())
 
         _LOGGER.info("started successfully in approx %.2f seconds", time.monotonic() - start_time)
@@ -1364,16 +1225,6 @@ async def request_guild_members(
             guild=guild, include_presences=include_presences, query=query, limit=limit, users=users, nonce=nonce
         )
 
-    async def _set_close_flag(self, signame: str, signum: int) -> None:
-        # This needs to be a coroutine, as the closing event is not threadsafe, so we have no way to set this
-        # from a Unix system call handler if we are running on a thread that isn't the main application thread
-        # without getting undefined behaviour. We do however have `asyncio.run_coroutine_threadsafe` which can
-        # run a coroutine function on the event loop from a completely different thread, so this is the safest
-        # solution.
-        _LOGGER.debug("received interrupt %s (%s), will start shutting down shortly", signame, signum)
-
-        await self._close()
-
     async def _start_one_shard(
         self,
         activity: typing.Optional[presences.Activity],
@@ -1384,8 +1235,7 @@ async def _start_one_shard(
         shard_id: int,
         shard_count: int,
         url: str,
-        closing_event: asyncio.Event,
-    ) -> shard_impl.GatewayShardImpl:
+    ) -> None:
         new_shard = shard_impl.GatewayShardImpl(
             http_settings=self._http_settings,
             proxy_settings=self._proxy_settings,
@@ -1402,14 +1252,20 @@ async def _start_one_shard(
             token=self._token,
             url=url,
         )
-        self._shards[shard_id] = new_shard
+        try:
+            start = time.monotonic()
+            await new_shard.start()
+            end = time.monotonic()
+
+            if new_shard.is_alive:
+                _LOGGER.debug("shard %s started successfully in %.1fms", shard_id, (end - start) * 1_000)
+                self._shards[shard_id] = new_shard
+                return
 
-        start = time.monotonic()
-        await aio.first_completed(new_shard.start(), closing_event.wait())
-        end = time.monotonic()
+            raise RuntimeError(f"shard {shard_id} shut down immediately when starting")
 
-        if new_shard.is_alive:
-            _LOGGER.debug("shard %s started successfully in %.1fms", shard_id, (end - start) * 1_000)
-            return new_shard
+        except Exception:
+            if new_shard.is_alive:
+                await new_shard.close()
 
-        raise errors.GatewayError(f"shard {shard_id} shut down immediately when starting")
+            raise
diff --git a/hikari/impl/buckets.py b/hikari/impl/buckets.py
index 1733c5dd..806020e6 100644
--- a/hikari/impl/buckets.py
+++ b/hikari/impl/buckets.py
@@ -395,9 +395,6 @@ def __exit__(
     ) -> None:
         self.close()
 
-    def __del__(self) -> None:
-        self.close()
-
     def start(self, poll_period: float = 20.0, expire_after: float = 10.0) -> None:
         """Start this ratelimiter up.
 
diff --git a/hikari/impl/config.py b/hikari/impl/config.py
index 581a878c..bafd4858 100644
--- a/hikari/impl/config.py
+++ b/hikari/impl/config.py
@@ -274,7 +274,7 @@ class HTTPSettings(config.HTTPSettings):
     """
 
     force_close_transports: bool = attr.field(default=True, validator=attr.validators.instance_of(bool))
-    """Toggle whether to force close transports on shutdown.
+    """Toggle whether to force close transports on shut down.
 
     This defaults to `builtins.True` to combat various protocol and asyncio
     issues present when using Microsoft Windows. If you are sure you know
diff --git a/hikari/impl/event_manager_base.py b/hikari/impl/event_manager_base.py
index e855d314..5fcdb11f 100644
--- a/hikari/impl/event_manager_base.py
+++ b/hikari/impl/event_manager_base.py
@@ -494,7 +494,7 @@ def get_listeners(
         if items := self._listeners.get(event_type):
             return items.copy()
 
-        return []
+        return ()
 
     # Yes, this is not generic. The reason for this is MyPy complains about
     # using ABCs that are not concrete in generic types passed to functions.
diff --git a/hikari/impl/interaction_server.py b/hikari/impl/interaction_server.py
index aac48123..553d90f2 100644
--- a/hikari/impl/interaction_server.py
+++ b/hikari/impl/interaction_server.py
@@ -27,7 +27,6 @@
 
 import asyncio
 import logging
-import threading
 import typing
 
 import aiohttp
@@ -358,13 +357,14 @@ async def close(self) -> None:
             return
 
         self._is_closing = True
-        self._application_fetch_lock = None
-        # This shutdown then cleanup ordering matters.
+        # This shut down then cleanup ordering matters.
         await self._server.shutdown()
         await self._server.cleanup()
+        self._server = None
+        self._application_fetch_lock = None
         self._close_event.set()
         self._close_event = None
-        self._server = None
+        self._is_closing = False
 
     async def join(self) -> None:
         """Wait for the process to halt before continuing."""
@@ -457,7 +457,6 @@ async def on_interaction(self, body: bytes, signature: bytes, timestamp: bytes)
     async def start(
         self,
         backlog: int = 128,
-        enable_signal_handlers: typing.Optional[bool] = None,
         host: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None,
         port: typing.Optional[int] = None,
         path: typing.Optional[str] = None,
@@ -474,16 +473,6 @@ async def start(
         backlog : builtins.int
             The number of unaccepted connections that the system will allow before
             refusing new connections.
-        enable_signal_handlers : typing.Optional[builtins.bool]
-            Defaults to `builtins.True` if this is started in the main thread.
-
-            If on a __non-Windows__ OS with builtin support for kernel-level
-            POSIX signals, then setting this to `builtins.True` will allow
-            treating keyboard interrupts and other OS signals to safely shut
-            down the application as calls to shut down the application properly
-            rather than just killing the process in a dirty state immediately.
-            You should leave this enabled unless you plan to implement your own
-            signal handling yourself.
         host : typing.Optional[typing.Union[builtins.str, aiohttp.web.HostSequence]]
             TCP/IP host or a sequence of hosts for the HTTP server.
         port : typing.Optional[builtins.int]
@@ -499,7 +488,7 @@ async def start(
         socket : typing.Optional[socket.socket]
             A pre-existing socket object to accept connections on.
         shutdown_timeout : builtins.float
-            A delay to wait for graceful server shutdown before forcefully
+            A delay to wait for graceful server shut down before forcefully
             disconnecting all open client sockets. This defaults to 60 seconds.
         ssl_context : typing.Optional[ssl.SSLContext]
             SSL context for HTTPS servers.
@@ -511,27 +500,26 @@ async def start(
         if self._server:
             raise errors.ComponentStateConflictError("Cannot start an already active interaction server")
 
-        if enable_signal_handlers is None:
-            enable_signal_handlers = threading.current_thread() is threading.main_thread()
-
         self._close_event = asyncio.Event()
         self._is_closing = False
+
         aio_app = aiohttp.web.Application()
         aio_app.add_routes([aiohttp.web.post("/", self.aiohttp_hook)])
-        self._server = aiohttp.web_runner.AppRunner(aio_app, handle_signals=enable_signal_handlers, access_log=_LOGGER)
+        self._server = aiohttp.web_runner.AppRunner(aio_app, access_log=_LOGGER)
         await self._server.setup()
+
         sites: typing.List[aiohttp.web.BaseSite] = []
 
         if host is not None:
             if isinstance(host, str):
-                host = [host]
+                host = (host,)
 
             for h in host:
                 sites.append(
                     aiohttp.web.TCPSite(
                         self._server,
                         h,
-                        port,
+                        port=port,
                         shutdown_timeout=shutdown_timeout,
                         ssl_context=ssl_context,
                         backlog=backlog,
@@ -540,7 +528,7 @@ async def start(
                     )
                 )
 
-        elif path is None and socket is None or port is None:
+        elif path is None and socket is None or port is not None:
             sites.append(
                 aiohttp.web.TCPSite(
                     self._server,
diff --git a/hikari/impl/rest_bot.py b/hikari/impl/rest_bot.py
index efba07f5..776fb6ca 100644
--- a/hikari/impl/rest_bot.py
+++ b/hikari/impl/rest_bot.py
@@ -38,6 +38,7 @@
 from hikari.impl import interaction_server as interaction_server_impl
 from hikari.impl import rest as rest_impl
 from hikari.internal import aio
+from hikari.internal import signals
 from hikari.internal import ux
 
 if typing.TYPE_CHECKING:
@@ -280,7 +281,7 @@ def __init__(
             token_type=token_type,
         )
 
-        # IntegrationServer
+        # InteractionServer
         self._server = interaction_server_impl.InteractionServer(
             entity_factory=self._entity_factory,
             public_key=public_key,
@@ -359,18 +360,22 @@ def print_banner(
 
     async def close(self) -> None:
         if not self._close_event:
-            raise errors.ComponentStateConflictError("Cannot close an inactive interaction server")
+            raise errors.ComponentStateConflictError("Cannot close an inactive bot")
 
         if self._is_closing:
             await self.join()
             return
 
+        _LOGGER.info("bot requested to shut down")
+
         self._is_closing = True
-        close_event = self._close_event
         await self._server.close()
         await self._rest.close()
-        close_event.set()
+        self._close_event.set()
         self._close_event = None
+        self._is_closing = False
+
+        _LOGGER.info("bot shut down successfully")
 
     async def join(self) -> None:
         if not self._close_event:
@@ -393,6 +398,7 @@ def run(
         host: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None,
         path: typing.Optional[str] = None,
         port: typing.Optional[int] = None,
+        propagate_interrupts: bool = False,
         reuse_address: typing.Optional[bool] = None,
         reuse_port: typing.Optional[bool] = None,
         shutdown_timeout: float = 60.0,
@@ -415,7 +421,7 @@ def run(
         close_loop : builtins.bool
             Defaults to `builtins.True`. If `builtins.True`, then once the bot
             enters a state where all components have shut down permanently
-            during application shutdown, then all asyncgens and background tasks
+            during application shut down, then all asyncgens and background tasks
             will be destroyed, and the event loop will be shut down.
 
             This will wait until all `hikari`-owned `aiohttp` connectors have
@@ -460,7 +466,7 @@ def run(
         socket : typing.Optional[socket.socket]
             A pre-existing socket object to accept connections on.
         shutdown_timeout : builtins.float
-            A delay to wait for graceful server shutdown before forcefully
+            A delay to wait for graceful server shut down before forcefully
             disconnecting all open client sockets. This defaults to 60 seconds.
         ssl_context : typing.Optional[ssl.SSLContext]
             SSL context for HTTPS servers.
@@ -480,36 +486,48 @@ def run(
                 _LOGGER.log(ux.TRACE, "cannot set coroutine tracking depth for sys, no functionality exists for this")
 
         try:
-            loop.run_until_complete(
-                self.start(
-                    backlog=backlog,
-                    check_for_updates=check_for_updates,
-                    enable_signal_handlers=enable_signal_handlers,
-                    host=host,
-                    port=port,
-                    path=path,
-                    reuse_address=reuse_address,
-                    reuse_port=reuse_port,
-                    socket=socket,
-                    shutdown_timeout=shutdown_timeout,
-                    ssl_context=ssl_context,
+            with signals.handle_interrupts(
+                enabled=enable_signal_handlers,
+                loop=loop,
+                propagate_interrupts=propagate_interrupts,
+            ):
+                loop.run_until_complete(
+                    self.start(
+                        backlog=backlog,
+                        check_for_updates=check_for_updates,
+                        host=host,
+                        port=port,
+                        path=path,
+                        reuse_address=reuse_address,
+                        reuse_port=reuse_port,
+                        socket=socket,
+                        shutdown_timeout=shutdown_timeout,
+                        ssl_context=ssl_context,
+                    )
                 )
-            )
-            loop.run_until_complete(self.join())
+                loop.run_until_complete(self.join())
 
         finally:
+            if self._close_event:
+                if self._is_closing:
+                    loop.run_until_complete(self._close_event.wait())
+                else:
+                    loop.run_until_complete(self.close())
+
             if close_passed_executor and self._executor:
+                _LOGGER.debug("shutting down executor %s", self._executor)
                 self._executor.shutdown(wait=True)
                 self._executor = None
 
             if close_loop:
-                loop.close()
+                aio.destroy_loop(loop, _LOGGER)
+
+            _LOGGER.info("successfully terminated")
 
     async def start(
         self,
         backlog: int = 128,
         check_for_updates: bool = True,
-        enable_signal_handlers: typing.Optional[bool] = None,
         host: typing.Optional[typing.Union[str, typing.Sequence[str]]] = None,
         port: typing.Optional[int] = None,
         path: typing.Optional[str] = None,
@@ -529,16 +547,6 @@ async def start(
         check_for_updates : builtins.bool
             Defaults to `builtins.True`. If `builtins.True`, will check for
             newer versions of `hikari` on PyPI and notify if available.
-        enable_signal_handlers : typing.Optional[builtins.bool]
-            Defaults to `builtins.True` if this is started in the main thread.
-
-            If on a __non-Windows__ OS with builtin support for kernel-level
-            POSIX signals, then setting this to `builtins.True` will allow
-            treating keyboard interrupts and other OS signals to safely shut
-            down the application as calls to shut down the application properly
-            rather than just killing the process in a dirty state immediately.
-            You should leave this enabled unless you plan to implement your own
-            signal handling yourself.
         host : typing.Optional[typing.Union[builtins.str, aiohttp.web.HostSequence]]
             TCP/IP host or a sequence of hosts for the HTTP server.
         port : typing.Optional[builtins.int]
@@ -554,7 +562,7 @@ async def start(
         socket : typing.Optional[socket.socket]
             A pre-existing socket object to accept connections on.
         shutdown_timeout : builtins.float
-            A delay to wait for graceful server shutdown before forcefully
+            A delay to wait for graceful server shut down before forcefully
             disconnecting all open client sockets. This defaults to 60 seconds.
         ssl_context : typing.Optional[ssl.SSLContext]
             SSL context for HTTPS servers.
@@ -578,7 +586,6 @@ async def start(
         self._rest.start()
         await self._server.start(
             backlog=backlog,
-            enable_signal_handlers=enable_signal_handlers,
             host=host,
             port=port,
             path=path,
diff --git a/hikari/impl/shard.py b/hikari/impl/shard.py
index 0a74d921..33bc26c3 100644
--- a/hikari/impl/shard.py
+++ b/hikari/impl/shard.py
@@ -45,6 +45,7 @@
 from hikari import undefined
 from hikari.api import shard
 from hikari.impl import rate_limits
+from hikari.internal import aio
 from hikari.internal import data_binding
 from hikari.internal import net
 from hikari.internal import time
@@ -81,11 +82,13 @@
 _INVALID_SESSION: typing.Final[int] = 9
 _HELLO: typing.Final[int] = 10
 _HEARTBEAT_ACK: typing.Final[int] = 11
+# Special dispatches
+_READY: typing.Final[str] = sys.intern("READY")
+_RESUMED: typing.Final[str] = sys.intern("RESUMED")
 # If we disconnect within this period of time after starting, we should
 # use an exponential backoff before restarting.
 _BACKOFF_WINDOW: typing.Final[float] = 30.0
 _BACKOFF_BASE: typing.Final[float] = 1.85
-_BACKOFF_INCREMENT_START: typing.Final[int] = 2
 _BACKOFF_CAP: typing.Final[float] = 60.0
 # Discord seems to invalidate sessions if I send a 1xxx, which is useless
 # for invalid session and reconnect messages where I want to be able to
@@ -93,11 +96,12 @@
 _RESUME_CLOSE_CODE: typing.Final[int] = 3_000
 # Per-shard sending rate-limit
 _TOTAL_RATELIMIT: typing.Final[typing.Tuple[float, int]] = (60.0, 120)
-# Rate-limit for chunking requests (used to prevent saturating the entire
-# ratelimit window).
-_CHUNKING_RATELIMIT: typing.Final[typing.Tuple[float, int]] = (60.0, 60)
+# Rate-limit for non-priority packages.
+# This is done to always allow for HEARTBEAT packages
+# to get around (leaving 3 slots for it).
+_NON_PRIORITY_RATELIMIT: typing.Final[typing.Tuple[float, int]] = (60.0, 117)
 # Supported gateway version
-_VERSION: int = 8
+_VERSION: typing.Final[int] = 8
 # Used to identify the end of a ZLIB payload
 _ZLIB_SUFFIX: typing.Final[bytes] = b"\x00\x00\xff\xff"
 # Close codes which don't invalidate the current session.
@@ -119,14 +123,8 @@ def filterer(entry: str) -> str:
     return filterer
 
 
-if typing.TYPE_CHECKING:
-    # noinspection PyProtectedMember,PyUnresolvedReferences
-    _ZlibDecompressor = zlib._Decompress
-
-
-# aiohttp.ClientWebSocketResponse isn't slotted
 @typing.final
-class _GatewayTransport(aiohttp.ClientWebSocketResponse):
+class _GatewayTransport:
     """Internal component to handle lower-level communication logic.
 
     This includes translating aiohttp error conditions to hikari ones,
@@ -136,65 +134,87 @@ class _GatewayTransport(aiohttp.ClientWebSocketResponse):
     Payload logging is also performed here.
     """
 
-    def __init__(self, *args: typing.Any, **kwargs: typing.Any) -> None:
-        super().__init__(*args, **kwargs)
-        self.zlib: _ZlibDecompressor = zlib.decompressobj()
-        self.sent_close = False
-
-        # Initialized from `connect'
-        # These are type-hinted here instead of above to prevent MyPy from misinterpreting typing.Callable as a method
-        self.logger: logging.Logger
-        self.log_filterer: typing.Callable[[str], str]
-
-    async def send_close(self, *, code: int = 1000, message: bytes = b"") -> None:
-        # aiohttp may close the socket by invoking close() internally. By giving
-        # a different name, we can ensure aiohttp won't invoke this method.
-        # We can then guarantee any call to this method was made by us, as
-        # opposed to, for example, Windows injecting a spurious EOF when
-        # something disconnects, which makes aiohttp just shut down as if we
-        # did it.
-        if self.sent_close:
+    __slots__ = (
+        "_zlib",
+        "_sent_close",
+        "_logger",
+        "_exit_stack",
+        "_log_filterer",
+        "_ws",
+        "_receive_and_check",
+        "_log_payload",
+    )
+
+    def __init__(
+        self,
+        ws: aiohttp.ClientWebSocketResponse,
+        transport_compression: bool,
+        exit_stack: contextlib.AsyncExitStack,
+        logger: logging.Logger,
+        log_filterer: typing.Callable[[str], str],
+    ) -> None:
+        self._logger = logger
+        self._log_payload = self._logger.isEnabledFor(ux.TRACE)
+        self._log_filterer = log_filterer
+        self._exit_stack = exit_stack
+        self._sent_close = False
+        self._ws = ws
+        self._zlib = zlib.decompressobj()
+
+        if transport_compression:
+            self._receive_and_check = self._receive_and_check_zlib
+        else:
+            self._receive_and_check = self._receive_and_check_text
+
+    async def send_close(self, *, code: int, message: bytes) -> None:
+        if self._sent_close:
             return
 
-        self.sent_close = True
-        self.logger.debug("sending close frame with code %s and message %s", int(code), message)
+        self._sent_close = True
+        self._logger.debug("sending close frame with code %s and message %s", code, message)
         try:
-            await asyncio.wait_for(super().close(code=code, message=message), timeout=5)
+            await asyncio.wait_for(self._ws.close(code=code, message=message), timeout=5)
+
         except asyncio.TimeoutError:
-            self.logger.debug("failed to send close frame in time, probably connection issues")
+            self._logger.debug("failed to send close frame in time, probably connection issues")
 
-    async def receive_json(
-        self,
-        *,
-        loads: aiohttp.typedefs.JSONDecoder = data_binding.load_json,
-        timeout: typing.Optional[float] = None,
-    ) -> typing.Any:
-        pl = await self._receive_and_check(timeout)
-        if self.logger.isEnabledFor(ux.TRACE):
-            filtered = self.log_filterer(pl)
-            self.logger.log(ux.TRACE, "received payload with size %s\n    %s", len(pl), filtered)
-        return loads(pl)
-
-    async def send_json(
-        self,
-        data: data_binding.JSONObject,
-        compress: typing.Optional[int] = None,
-        *,
-        dumps: aiohttp.typedefs.JSONEncoder = data_binding.dump_json,
-    ) -> None:
-        pl = dumps(data)
-        if self.logger.isEnabledFor(ux.TRACE):
-            filtered = self.log_filterer(pl)
-            self.logger.log(ux.TRACE, "sending payload with size %s\n    %s", len(pl), filtered)
-        await self.send_str(pl, compress)
+        finally:
+            await self._exit_stack.aclose()
+
+            # We have to sleep to allow aiohttp time to close SSL transports...
+            # This code can be removed in aiohttp v4.0.0
+            # https://github.com/aio-libs/aiohttp/issues/1925
+            # https://docs.aiohttp.org/en/stable/client_advanced.html#graceful-shutdown
+            await asyncio.sleep(0.25)
+
+    async def receive_json(self) -> typing.Any:
+        pl = await self._receive_and_check()
+        if self._log_payload:
+            filtered = self._log_filterer(pl)
+            self._logger.log(ux.TRACE, "received payload with size %s\n    %s", len(pl), filtered)
+
+        return data_binding.load_json(pl)
+
+    async def send_json(self, data: data_binding.JSONObject) -> None:
+        pl = data_binding.dump_json(data)
+        if self._log_payload:
+            filtered = self._log_filterer(pl)
+            self._logger.log(ux.TRACE, "sending payload with size %s\n    %s", len(pl), filtered)
+
+        await self._ws.send_str(pl)
 
     def _handle_other_message(self, message: aiohttp.WSMessage, /) -> typing.NoReturn:
+        if message.type == aiohttp.WSMsgType.TEXT:
+            raise errors.GatewayError("Unexpected message type received TEXT, expected BINARY")
+
+        if message.type == aiohttp.WSMsgType.BINARY:
+            raise errors.GatewayError("Unexpected message type received BINARY, expected TEXT")
+
         if message.type == aiohttp.WSMsgType.CLOSE:
             close_code = int(message.data)
-            reason = message.extra
-            self.logger.error("connection closed with code %s (%s)", close_code, reason)
+
             can_reconnect = close_code < 4000 or close_code in _RECONNECTABLE_CLOSE_CODES
-            raise errors.GatewayServerClosedConnectionError(reason, close_code, can_reconnect)
+            raise errors.GatewayServerClosedConnectionError(message.extra, close_code, can_reconnect)
 
         if message.type == aiohttp.WSMsgType.CLOSING or message.type == aiohttp.WSMsgType.CLOSED:
             # May be caused by the server shutting us down.
@@ -203,50 +223,43 @@ def _handle_other_message(self, message: aiohttp.WSMessage, /) -> typing.NoRetur
             raise errors.GatewayConnectionError("Socket has closed")
 
         # Assume exception for now.
-        ex = self.exception()
-        self.logger.warning(
-            "encountered unexpected error: %s",
-            ex,
-            exc_info=ex if self.logger.isEnabledFor(logging.DEBUG) else None,
-        )
-        raise errors.GatewayError("Unexpected websocket exception from gateway") from ex
+        raise errors.GatewayError("Unexpected websocket exception from gateway") from self._ws.exception()
 
-    async def _receive_and_check(self, timeout: typing.Optional[float], /) -> str:
-        message = await self.receive(timeout)
+    async def _receive_and_check_text(self) -> str:
+        message = await self._ws.receive()
 
         if message.type == aiohttp.WSMsgType.TEXT:
             assert isinstance(message.data, str)
             return message.data
 
+        self._handle_other_message(message)
+
+    async def _receive_and_check_zlib(self) -> str:
+        message = await self._ws.receive()
+
         if message.type == aiohttp.WSMsgType.BINARY:
             if message.data.endswith(_ZLIB_SUFFIX):
-                return self.zlib.decompress(message.data).decode("utf-8")
+                return self._zlib.decompress(message.data).decode("utf-8")
 
-            return await self._receive_and_check_complete_zlib_package(message.data, timeout)
+            return await self._receive_and_check_complete_zlib_package(message.data)
 
         self._handle_other_message(message)
 
-    async def _receive_and_check_complete_zlib_package(
-        self, initial_data: bytes, timeout: typing.Optional[float], /
-    ) -> str:
+    async def _receive_and_check_complete_zlib_package(self, initial_data: bytes, /) -> str:
         buff = bytearray(initial_data)
 
         while not buff.endswith(_ZLIB_SUFFIX):
-            message = await self.receive(timeout)
+            message = await self._ws.receive()
 
             if message.type == aiohttp.WSMsgType.BINARY:
                 buff.extend(message.data)
                 continue
 
-            if message.type == aiohttp.WSMsgType.TEXT:
-                raise errors.GatewayError("Unexpected message type received TEXT, expected BINARY")
-
             self._handle_other_message(message)
 
-        return self.zlib.decompress(buff).decode("utf-8")
+        return self._zlib.decompress(buff).decode("utf-8")
 
     @classmethod
-    @contextlib.asynccontextmanager
     async def connect(
         cls,
         *,
@@ -254,97 +267,86 @@ async def connect(
         logger: logging.Logger,
         proxy_settings: config.ProxySettings,
         log_filterer: typing.Callable[[str], str],
+        transport_compression: bool,
         url: str,
-    ) -> typing.AsyncGenerator[_GatewayTransport, None]:
+    ) -> _GatewayTransport:
         """Generate a single-use websocket connection.
 
         This uses a single connection in a TCP connector pool, with a one-use
         aiohttp client session.
-
-        This also handles waiting for transports to be closed properly first,
-        and keeps all of the nested boilerplate out of the way of the
-        rest of the code, for the most part anyway.
         """
         exit_stack = contextlib.AsyncExitStack()
 
         try:
-            connector = net.create_tcp_connector(http_settings, dns_cache=False, limit=1)
-            client_session = await exit_stack.enter_async_context(
-                net.create_client_session(connector, True, http_settings, True, proxy_settings.trust_env, cls)
-            )
-
-            web_socket = await exit_stack.enter_async_context(
-                client_session.ws_connect(
-                    max_msg_size=0,
-                    proxy=proxy_settings.url,
-                    proxy_headers=proxy_settings.headers,
-                    url=url,
-                    # We manage these ourselves
-                    autoclose=False,
-                )
-            )
-            assert isinstance(web_socket, cls)
-
-            web_socket.logger = logger
-            # We store this so we can remove it from debug logs
-            # which enables people to send logs in issues safely.
-            web_socket.log_filterer = log_filterer
-
-            raised = False
             try:
-                yield web_socket
-            except errors.GatewayError:
-                raised = True
-                raise
-            except Exception as ex:
-                raised = True
-                raise errors.GatewayError(f"Unexpected {type(ex).__name__}: {ex}") from ex
-            finally:
-                if web_socket.closed:
-                    logger.log(ux.TRACE, "ws was already closed")
-
-                elif raised:
-                    await web_socket.send_close(
-                        code=errors.ShardCloseCode.UNEXPECTED_CONDITION,
-                        message=b"unexpected fatal client error :-(",
+                connector = net.create_tcp_connector(http_settings=http_settings, dns_cache=False, limit=1)
+                client_session = await exit_stack.enter_async_context(
+                    net.create_client_session(
+                        connector=connector,
+                        connector_owner=True,
+                        http_settings=http_settings,
+                        raise_for_status=True,
+                        trust_env=proxy_settings.trust_env,
                     )
+                )
 
-                elif not web_socket._closing:
-                    # We use a special close code here that prevents Discord
-                    # randomly invalidating our session. Undocumented behaviour is
-                    # nice like that...
-                    await web_socket.send_close(
-                        code=_RESUME_CLOSE_CODE,
-                        message=b"client is shutting down",
+                web_socket = await exit_stack.enter_async_context(
+                    client_session.ws_connect(
+                        max_msg_size=0,
+                        proxy=proxy_settings.url,
+                        proxy_headers=proxy_settings.headers,
+                        url=url,
+                        # We manage this ourselves
+                        autoclose=False,
                     )
+                )
 
-        except (
-            aiohttp.ClientOSError,
-            aiohttp.ClientConnectionError,
-            aiohttp.WSServerHandshakeError,
-            asyncio.TimeoutError,
-        ) as ex:
-            # Windows will sometimes raise an aiohttp.ClientOSError
-            # If we cannot do DNS lookup, this will fail with a ClientConnectionError
-            # usually, but it might also fail with asyncio.TimeoutError if its gets stuck in a weird way
-            #
-            # aiohttp.WSServerHandshakeError has a really bad str, so we use the repr instead
-            if isinstance(ex, aiohttp.WSServerHandshakeError):
-                reason = repr(ex)
-            elif isinstance(ex, asyncio.TimeoutError):
-                reason = "Timeout exceeded"
-            else:
-                reason = str(ex)
-            raise errors.GatewayConnectionError(reason) from None
+                return cls(
+                    ws=web_socket,
+                    transport_compression=transport_compression,
+                    exit_stack=exit_stack,
+                    logger=logger,
+                    log_filterer=log_filterer,
+                )
 
-        finally:
+            except (aiohttp.ClientConnectionError, aiohttp.ClientResponseError, asyncio.TimeoutError) as ex:
+                # If we cannot do DNS lookup, this will fail with an aiohttp.ClientConnectionError
+                # usually, but it might also fail with asyncio.TimeoutError if its gets stuck in a weird way
+                #
+                # aiohttp.ClientResponseError has a really bad str, so we use the repr instead
+                if isinstance(ex, aiohttp.ClientResponseError):
+                    reason = repr(ex)
+                elif isinstance(ex, asyncio.TimeoutError):
+                    reason = "Timeout exceeded"
+                else:
+                    reason = str(ex)
+                raise errors.GatewayConnectionError(reason) from None
+
+        except Exception:
             await exit_stack.aclose()
 
             # We have to sleep to allow aiohttp time to close SSL transports...
+            # This code can be removed in aiohttp v4.0.0
             # https://github.com/aio-libs/aiohttp/issues/1925
             # https://docs.aiohttp.org/en/stable/client_advanced.html#graceful-shutdown
             await asyncio.sleep(0.25)
 
+            raise
+
+
+def _serialize_datetime(dt: typing.Optional[datetime.datetime]) -> typing.Optional[int]:
+    if dt is None:
+        return None
+
+    return int(dt.timestamp() * 1_000)
+
+
+def _serialize_activity(activity: typing.Optional[presences.Activity]) -> data_binding.JSONish:
+    if activity is None:
+        return None
+
+    return {"name": activity.name, "type": int(activity.type), "url": activity.url}
+
 
 @typing.final
 class GatewayShardImpl(shard.GatewayShard):
@@ -410,25 +412,24 @@ class GatewayShardImpl(shard.GatewayShard):
 
     __slots__: typing.Sequence[str] = (
         "_activity",
-        "_closed_event",
-        "_closing_event",
-        "_chunking_rate_limit",
         "_event_manager",
         "_event_factory",
         "_gateway_url",
-        "_handshake_completed",
+        "_handshake_event",
         "_heartbeat_latency",
         "_http_settings",
         "_idle_since",
         "_intents",
         "_is_afk",
+        "_is_closing",
+        "_keep_alive_task",
         "_large_threshold",
         "_last_heartbeat_ack_received",
         "_last_heartbeat_sent",
         "_logger",
+        "_non_priority_rate_limit",
         "_proxy_settings",
         "_resume_gateway_url",
-        "_run_task",
         "_seq",
         "_session_id",
         "_shard_count",
@@ -461,7 +462,6 @@ def __init__(
         token: str,
         url: str,
     ) -> None:
-
         if data_format != shard.GatewayDataFormat.JSON:
             raise NotImplementedError(f"Unsupported gateway data format: {data_format}")
 
@@ -469,28 +469,26 @@ def __init__(
             raise NotImplementedError(f"Unsupported compression format {compression}")
 
         self._activity = initial_activity
-        self._closing_event = asyncio.Event()
-        self._closed_event = asyncio.Event()
-        self._chunking_rate_limit = rate_limits.WindowedBurstRateLimiter(
-            f"shard {shard_id} chunking rate limit",
-            *_CHUNKING_RATELIMIT,
-        )
         self._event_manager = event_manager
         self._event_factory = event_factory
         self._gateway_url = url
-        self._handshake_completed = asyncio.Event()
+        self._handshake_event: typing.Optional[asyncio.Event] = None
         self._heartbeat_latency = float("nan")
         self._http_settings = http_settings
         self._idle_since = initial_idle_since
         self._intents = intents
         self._is_afk = initial_is_afk
+        self._is_closing = False
+        self._keep_alive_task: typing.Optional[asyncio.Task[None]] = None
         self._large_threshold = large_threshold
         self._last_heartbeat_ack_received = float("nan")
         self._last_heartbeat_sent = float("nan")
         self._logger = logging.getLogger(f"hikari.gateway.{shard_id}")
+        self._non_priority_rate_limit = rate_limits.WindowedBurstRateLimiter(
+            f"shard {shard_id} non-priority rate limit", *_NON_PRIORITY_RATELIMIT
+        )
         self._proxy_settings = proxy_settings
         self._resume_gateway_url: typing.Optional[str] = None
-        self._run_task: typing.Optional[asyncio.Task[None]] = None
         self._seq: typing.Optional[int] = None
         self._session_id: typing.Optional[str] = None
         self._shard_count = shard_count
@@ -498,8 +496,7 @@ def __init__(
         self._status = initial_status
         self._token = token
         self._total_rate_limit = rate_limits.WindowedBurstRateLimiter(
-            f"shard {shard_id} total rate limit",
-            *_TOTAL_RATELIMIT,
+            f"shard {shard_id} total rate limit", *_TOTAL_RATELIMIT
         )
         self._transport_compression = compression is not None
         self._user_id: typing.Optional[snowflakes.Snowflake] = None
@@ -519,61 +516,63 @@ def intents(self) -> intents_.Intents:
 
     @property
     def is_alive(self) -> bool:
-        return self._ws is not None and not self._ws.sent_close
+        return self._keep_alive_task is not None
+
+    @property
+    def is_connected(self) -> bool:
+        return self._ws is not None and self._handshake_event is not None and self._handshake_event.is_set()
 
     @property
     def shard_count(self) -> int:
         return self._shard_count
 
     async def close(self) -> None:
-        self._check_if_alive()
-        if not self._closing_event.is_set():
-            try:
-                if self._ws is not None:
-                    self._logger.debug(
-                        "shard.close() was called and the websocket was still alive -- "
-                        "disconnecting immediately with GOING AWAY"
-                    )
-                    await self._ws.send_close(code=errors.ShardCloseCode.GOING_AWAY, message=b"shard disconnecting")
-                self._closing_event.set()
-            finally:
-                self._chunking_rate_limit.close()
-                self._total_rate_limit.close()
+        if not self._keep_alive_task:
+            raise errors.ComponentStateConflictError("Cannot close an inactive shard")
+
+        if self._is_closing:
+            await self.join()
+            return
 
-        # Wait for the shard to fully close
-        await self._closed_event.wait()
+        self._logger.info("shard has been requested to shutdown")
+        self._is_closing = True
 
-    async def get_user_id(self) -> snowflakes.Snowflake:
-        await self._handshake_completed.wait()
-        if self._user_id is None:
-            raise RuntimeError("user_id was not known, this is probably a bug")
+        self._keep_alive_task.cancel()
+        try:
+            await self._keep_alive_task
+        except asyncio.CancelledError:
+            pass
+
+        self._keep_alive_task = None
+        self._non_priority_rate_limit.close()
+        self._total_rate_limit.close()
+        self._is_closing = False
+        self._logger.info("shard shutdown successfully")
+
+    def get_user_id(self) -> snowflakes.Snowflake:
+        self._check_if_connected()
+        assert self._user_id is not None, "user_id was not known, this is probably a bug"
         return self._user_id
 
-    def _get_ws(self) -> _GatewayTransport:
-        if not self._ws:
-            raise errors.ComponentStateConflictError("Shard is shutting down")
+    async def join(self) -> None:
+        if not self._keep_alive_task:
+            raise errors.ComponentStateConflictError("Cannot join an inactive shard")
 
-        return self._ws
+        await asyncio.wait_for(asyncio.shield(self._keep_alive_task), timeout=None)
 
-    async def join(self) -> None:
-        self._check_if_alive()
-        await self._closed_event.wait()
+    async def _send_json(self, data: data_binding.JSONObject, /, priority: bool = False) -> None:
+        if not priority:
+            await self._non_priority_rate_limit.acquire()
 
-    async def _send_json(
-        self,
-        data: data_binding.JSONObject,
-        compress: typing.Optional[int] = None,
-        *,
-        dumps: aiohttp.typedefs.JSONEncoder = data_binding.dump_json,
-    ) -> None:
         await self._total_rate_limit.acquire()
 
-        await self._get_ws().send_json(data=data, compress=compress, dumps=dumps)
+        assert self._ws is not None
+        await self._ws.send_json(data)
 
-    def _check_if_alive(self) -> None:
-        if not self.is_alive:
+    def _check_if_connected(self) -> None:
+        if not self.is_connected:
             raise errors.ComponentStateConflictError(
-                f"shard {self._shard_id} is not running so it cannot be interacted with"
+                f"shard {self._shard_id} is not connected so it cannot be interacted with"
             )
 
     async def request_guild_members(
@@ -586,7 +585,7 @@ async def request_guild_members(
         users: undefined.UndefinedOr[snowflakes.SnowflakeishSequence[users_.User]] = undefined.UNDEFINED,
         nonce: undefined.UndefinedOr[str] = undefined.UNDEFINED,
     ) -> None:
-        self._check_if_alive()
+        self._check_if_connected()
         if not query and not limit and not self._intents & intents_.Intents.GUILD_MEMBERS:
             raise errors.MissingIntentError(intents_.Intents.GUILD_MEMBERS)
 
@@ -605,8 +604,6 @@ async def request_guild_members(
         if nonce is not undefined.UNDEFINED and len(bytes(nonce, "utf-8")) > 32:
             raise ValueError("'nonce' can be no longer than 32 byte characters long.")
 
-        await self._chunking_rate_limit.acquire()
-
         payload = data_binding.JSONObjectBuilder()
         payload.put_snowflake("guild_id", guild)
         payload.put("presences", include_presences)
@@ -618,22 +615,25 @@ async def request_guild_members(
         await self._send_json({_OP: _REQUEST_GUILD_MEMBERS, _D: payload})
 
     async def start(self) -> None:
-        if self._run_task is not None:
+        if self._keep_alive_task or self._handshake_event:
             raise errors.ComponentStateConflictError("Cannot run more than one instance of one shard concurrently")
 
-        run_task = asyncio.create_task(self._run(), name=f"run shard {self._shard_id}")
-        self._run_task = run_task
-        waiter = asyncio.create_task(self._handshake_completed.wait(), name=f"wait for shard {self._shard_id} to start")
-        done, _ = await asyncio.wait((waiter, run_task), return_when=asyncio.FIRST_COMPLETED)
-        waiter.cancel()
+        self._handshake_event = asyncio.Event()
+        keep_alive_task = asyncio.create_task(
+            self._keep_alive(),
+            name=f"keep alive (shard {self._shard_id})",
+        )
+
+        await aio.first_completed(self._handshake_event.wait(), asyncio.shield(keep_alive_task))
 
-        if done and waiter not in done:
+        if not self._handshake_event.is_set():
             # This might throw an error, or it might not, depending on what we do with it.
             # This occurs if the run task finished before the handshake completion event,
             # which implies the shard died before it could become ready/resume...
-            self._run_task = None
-            run_task.result()
-            raise asyncio.CancelledError(f"shard {self._shard_id} was closed before it could start successfully")
+            keep_alive_task.result()
+            raise RuntimeError(f"shard {self._shard_id} was closed before it could start successfully")
+
+        self._keep_alive_task = keep_alive_task
 
     async def update_presence(
         self,
@@ -643,15 +643,14 @@ async def update_presence(
         activity: undefined.UndefinedNoneOr[presences.Activity] = undefined.UNDEFINED,
         status: undefined.UndefinedOr[presences.Status] = undefined.UNDEFINED,
     ) -> None:
-        self._check_if_alive()
+        self._check_if_connected()
         presence_payload = self._serialize_and_store_presence_payload(
             idle_since=idle_since,
             afk=afk,
             activity=activity,
             status=status,
         )
-        payload: data_binding.JSONObject = {_OP: _PRESENCE_UPDATE, _D: presence_payload}
-        await self._send_json(payload)
+        await self._send_json({_OP: _PRESENCE_UPDATE, _D: presence_payload})
 
     async def update_voice_state(
         self,
@@ -661,7 +660,7 @@ async def update_voice_state(
         self_mute: undefined.UndefinedOr[bool] = undefined.UNDEFINED,
         self_deaf: undefined.UndefinedOr[bool] = undefined.UNDEFINED,
     ) -> None:
-        self._check_if_alive()
+        self._check_if_connected()
 
         payload = data_binding.JSONObjectBuilder()
         payload.put_snowflake("guild_id", guild)
@@ -671,214 +670,105 @@ async def update_voice_state(
 
         await self._send_json({_OP: _VOICE_STATE_UPDATE, _D: payload})
 
-    def _dispatch(self, name: str, seq: int, data: data_binding.JSONObject) -> None:
-        # This is invoked a lot, and we don't need to explicitly await anything, so it should
-        # not be a coroutine. Makes event dispatches much much faster under significant load.
-
-        self._seq = seq
-
-        if name == "READY":
-            self._session_id = data["session_id"]
-            self._resume_gateway_url = data["resume_gateway_url"]
-            user_pl = data["user"]
-            user_id = user_pl["id"]
-            self._user_id = snowflakes.Snowflake(user_id)
-            tag = user_pl["username"] + "#" + user_pl["discriminator"]
-            unavailable_guild_count = len(data["guilds"])
-            version = data["v"]
-            self._logger.info(
-                "shard is ready: %s guilds, %s (%s), session %r on v%s gateway",
-                unavailable_guild_count,
-                tag,
-                user_id,
-                self._session_id,
-                version,
-            )
-            self._handshake_completed.set()
-
-        elif name == "RESUMED":
-            self._logger.info("shard has resumed [session:%s, seq:%s]", self._session_id, self._seq)
-            self._handshake_completed.set()
+    async def _send_heartbeat(self) -> None:
+        self._logger.log(ux.TRACE, "sending HEARTBEAT [s:%s]", self._seq)
+        await self._send_json({_OP: _HEARTBEAT, _D: self._seq}, priority=True)
+        self._last_heartbeat_sent = time.monotonic()
 
-        try:
-            self._event_manager.consume_raw_event(name, self, data)
-        except LookupError:
-            self._logger.debug("ignoring unknown event %s:\n    %r", name, data)
-
-    async def _identify(self) -> None:
-        payload: data_binding.JSONObject = {
-            _OP: _IDENTIFY,
-            _D: {
-                "token": self._token,
-                "compress": False,
-                "large_threshold": self._large_threshold,
-                "properties": {
-                    "os": f"{platform.system()} {platform.architecture()[0]}",
-                    "browser": f"hikari ({about.__version__}, aiohttp {aiohttp.__version__})",
-                    "device": f"hikari {about.__version__}",
-                },
-                "shard": [self._shard_id, self._shard_count],
-            },
-        }
-
-        payload[_D]["intents"] = self._intents
-
-        payload[_D]["presence"] = self._serialize_and_store_presence_payload()
-
-        await self._send_json(payload)
-
-    async def _heartbeat(self, heartbeat_interval: float) -> bool:
-        # Return True if zombied or should reconnect, false if time to die forever.
+    async def _heartbeat(self, heartbeat_interval: float) -> None:
         # Prevent immediately zombie-ing.
         self._last_heartbeat_ack_received = time.monotonic()
         self._logger.debug("starting heartbeat with interval %ss", heartbeat_interval)
 
-        while not self._closing_event.is_set() and not self._closed_event.is_set():
+        while True:
             if self._last_heartbeat_ack_received <= self._last_heartbeat_sent:
                 # Gateway is zombie, close and request reconnect.
-                self._logger.warning(
-                    "connection has not received a HEARTBEAT_ACK for approx %.1fs and is being disconnected, "
-                    "expect a reconnect shortly",
+                self._logger.error(
+                    "connection has not received a HEARTBEAT_ACK for approx %.1fs and is being disconnected; "
+                    "will attempt to reconnect",
                     time.monotonic() - self._last_heartbeat_ack_received,
                 )
-                return True
-
-            self._logger.log(
-                ux.TRACE, "preparing to send HEARTBEAT [s:%s, interval:%ss]", self._seq, heartbeat_interval
-            )
-
-            await self._send_heartbeat()
-
-            try:
-                await asyncio.wait_for(self._closing_event.wait(), timeout=heartbeat_interval)
-                # We are closing
-                break
-            except asyncio.TimeoutError:
-                # We should continue
-                continue
+                return
 
-        self._logger.debug("heartbeat task is finishing now")
-        return False
-
-    async def _poll_events(self) -> typing.Optional[bool]:
-        payload = await self._get_ws().receive_json(timeout=5)
-
-        op = payload[_OP]  # opcode int
-        d = payload[_D]  # data/payload. Usually a dict or a bool for INVALID_SESSION
-
-        if op == _DISPATCH:
-            t = payload[_T]  # event name str
-            s = payload[_S]  # seq int
-            self._logger.log(ux.TRACE, "dispatching %s with seq %s", t, s)
-            self._dispatch(t, s, d)
-        elif op == _HEARTBEAT:
             await self._send_heartbeat()
-            self._logger.log(ux.TRACE, "sent HEARTBEAT")
-        elif op == _HEARTBEAT_ACK:
-            now = time.monotonic()
-            self._last_heartbeat_ack_received = now
-            self._heartbeat_latency = now - self._last_heartbeat_sent
-            self._logger.log(ux.TRACE, "received HEARTBEAT ACK in %.1fms", self._heartbeat_latency * 1_000)
-        elif op == _RECONNECT:
-            # We should be able to resume...
-            self._logger.info("received instruction to reconnect, will resume existing session")
-            return True
-        elif op == _INVALID_SESSION:
-            # We can resume if the payload was `true`.
-            if not d:
-                self._logger.info("received invalid session, will need to start a new session")
-                self._seq = None
-                self._resume_gateway_url = None
-                self._session_id = None
-            else:
-                self._logger.info("received invalid session, will resume existing session")
-            return True
-        else:
-            self._logger.log(ux.TRACE, "unknown opcode %s received, it will be ignored...", op)
-
-        return None
-
-    async def _resume(self) -> None:
-        await self._send_json(
-            {
-                _OP: _RESUME,
-                _D: {"token": self._token, "seq": self._seq, "session_id": self._session_id},
-            }
-        )
-
-    async def _run(self) -> None:
-        self._closed_event.clear()
-        self._closing_event.clear()
-        last_started_at = -float("inf")
 
-        backoff = rate_limits.ExponentialBackOff(
-            base=_BACKOFF_BASE,
-            maximum=_BACKOFF_CAP,
-            initial_increment=_BACKOFF_INCREMENT_START,
-        )
+            await asyncio.sleep(heartbeat_interval)
 
-        try:
-            while not self._closing_event.is_set() and not self._closed_event.is_set():
-                if time.monotonic() - last_started_at < _BACKOFF_WINDOW:
-                    backoff_time = next(backoff)
-                    self._logger.info("backing off reconnecting for %.2fs", backoff_time)
-
-                    try:
-                        await asyncio.wait_for(self._closing_event.wait(), timeout=backoff_time)
-                        # We were told to close.
-                        return
-                    except asyncio.TimeoutError:
-                        # We are going to run once.
-                        pass
+    async def _poll_events(self) -> None:
+        assert self._ws is not None
+        assert self._handshake_event is not None
 
-                try:
-                    last_started_at = time.monotonic()
-                    should_restart = await self._run_once()
+        while True:
+            payload = await self._ws.receive_json()
 
-                    # Since nothing went wrong, we can reset the backoff
-                    backoff.reset()
+            op = payload[_OP]
 
-                    if not should_restart:
-                        self._logger.info("shard has disconnected and shut down normally")
-                        return
+            if op == _DISPATCH:
+                name = payload[_T]
+                data = payload[_D]
+                self._seq = payload[_S]
 
-                except errors.GatewayConnectionError as ex:
-                    self._logger.warning(
-                        "failed to communicate with server, reason was: %r. Will retry shortly", ex.reason
-                    )
+                self._logger.log(ux.TRACE, "dispatching %s with seq %s", name, self._seq)
 
-                except errors.GatewayServerClosedConnectionError as ex:
-                    if not ex.can_reconnect:
-                        raise
+                if name == _READY:
+                    self._session_id = data["session_id"]
+                    self._resume_gateway_url = data["resume_gateway_url"]
+                    user_pl = data["user"]
+                    self._user_id = snowflakes.Snowflake(user_pl["id"])
 
                     self._logger.info(
-                        "server has closed connection, will reconnect if possible [code:%s, reason:%s]",
-                        ex.code,
-                        ex.reason,
+                        "shard is ready: %s guilds, %s (%s), session %r on v%s gateway",
+                        len(data["guilds"]),
+                        f"{user_pl['username']}#{user_pl['discriminator']}",
+                        self._user_id,
+                        self._session_id,
+                        data["v"],
                     )
+                    self._handshake_event.set()
 
-                    # We don't want to back off from this. If Discord keep closing the connection, it is their issue.
-                    # If we back off here, we'll find a mass outage will prevent shards from becoming healthy on
-                    # reconnect in large sharded bots for a very long period of time.
-                    backoff.reset()
+                elif name == _RESUMED:
+                    self._logger.info("resumed session [session:%s, seq:%s]", self._session_id, self._seq)
+                    self._handshake_event.set()
 
-                except errors.GatewayError as ex:
-                    self._logger.error("encountered generic gateway error", exc_info=ex)
-                    raise
+                try:
+                    self._event_manager.consume_raw_event(name, self, data)
+                except LookupError:
+                    self._logger.debug("ignoring unknown event %s:\n    %r", name, data)
+
+            elif op == _HEARTBEAT_ACK:
+                now = time.monotonic()
+                self._last_heartbeat_ack_received = now
+                self._heartbeat_latency = now - self._last_heartbeat_sent
+                self._logger.log(ux.TRACE, "received HEARTBEAT ACK in %.1fms", self._heartbeat_latency * 1_000)
+
+            elif op == _HEARTBEAT:
+                self._logger.log(ux.TRACE, "sending heartbeat as requested by gateway")
+                await self._send_heartbeat()
+
+            elif op == _RECONNECT:
+                self._logger.info("received instruction to reconnect, will resume existing session")
+                return
+
+            elif op == _INVALID_SESSION:
+                can_reconnect = payload[_D]  # We can resume if the payload data is `true`.
+                if not can_reconnect:
+                    self._logger.info("received invalid session, will need to start a new session")
+                    self._seq = None
+                    self._resume_gateway_url = None
+                    self._session_id = None
+                else:
+                    self._logger.info("received invalid session, will resume existing session")
 
-                except Exception as ex:
-                    self._logger.error("encountered some unhandled error", exc_info=ex)
-                    raise
-        finally:
-            self._closing_event.set()
-            self._closed_event.set()
+                return
 
-    async def _run_once(self) -> bool:
-        self._handshake_completed.clear()
-        dispatch_disconnect = False
-        exception = None
+            else:
+                self._logger.log(ux.TRACE, "unknown opcode %s received, it will be ignored...", op)
 
-        exit_stack = contextlib.AsyncExitStack()
+    async def _connect(self) -> typing.Tuple[asyncio.Task[None], ...]:
+        if self._ws is not None:
+            raise errors.ComponentStateConflictError("Attempting to connect an already connected shard")
+
+        assert self._handshake_event is not None
 
         url_parts = urllib.parse.urlparse(
             self._resume_gateway_url or self._gateway_url,
@@ -903,109 +793,156 @@ async def _run_once(self) -> bool:
             )
         )
 
-        self._ws = await exit_stack.enter_async_context(
-            _GatewayTransport.connect(
-                http_settings=self._http_settings,
-                log_filterer=_log_filterer(self._token),
-                logger=self._logger,
-                proxy_settings=self._proxy_settings,
-                url=url,
+        self._ws = await _GatewayTransport.connect(
+            http_settings=self._http_settings,
+            log_filterer=_log_filterer(self._token),
+            logger=self._logger,
+            proxy_settings=self._proxy_settings,
+            transport_compression=self._transport_compression,
+            url=url,
+        )
+        self._event_manager.dispatch(self._event_factory.deserialize_connected_event(self))
+
+        # Expect initial HELLO
+        hello_payload = await self._ws.receive_json()
+        if hello_payload[_OP] != _HELLO:
+            self._logger.debug(
+                "expected %s (HELLO) opcode, received %s which makes no sense, closing with PROTOCOL ERROR",
+                _HELLO,
+                hello_payload[_OP],
             )
+            await self._ws.send_close(code=errors.ShardCloseCode.PROTOCOL_ERROR, message=b"Expected HELLO op")
+            raise errors.GatewayError(f"Expected opcode {_HELLO} (HELLO), but received {hello_payload[_OP]}")
+
+        # Spawn lifetime tasks
+        heartbeat_interval = float(hello_payload[_D]["heartbeat_interval"]) / 1_000.0
+        heartbeat_task = asyncio.create_task(
+            self._heartbeat(heartbeat_interval), name=f"heartbeat (shard {self._shard_id})"
         )
+        poll_events_task = asyncio.create_task(self._poll_events(), name=f"poll events (shard {self._shard_id})")
+
+        # Perform handshake
+        if self._seq is None:
+            self._logger.debug("identifying with new session")
+            await self._send_json(
+                {
+                    _OP: _IDENTIFY,
+                    _D: {
+                        "token": self._token,
+                        "compress": False,
+                        "large_threshold": self._large_threshold,
+                        "properties": {
+                            "os": f"{platform.system()} {platform.architecture()[0]}",
+                            "browser": f"hikari ({about.__version__}, aiohttp {aiohttp.__version__})",
+                            "device": f"hikari {about.__version__}",
+                        },
+                        "shard": [self._shard_id, self._shard_count],
+                        "intents": self._intents,
+                        "presence": self._serialize_and_store_presence_payload(),
+                    },
+                }
+            )
+        else:
+            self._logger.debug("resuming session %s", self._session_id)
+            await self._send_json(
+                {
+                    _OP: _RESUME,
+                    _D: {"token": self._token, "seq": self._seq, "session_id": self._session_id},
+                }
+            )
 
-        try:
-            # Dispatch CONNECTED synthetic event.
-            self._event_manager.dispatch(self._event_factory.deserialize_connected_event(self))
-            dispatch_disconnect = True
+        lifetime_tasks = (heartbeat_task, poll_events_task)
 
-            heartbeat_task = await self._wait_for_hello()
+        await aio.first_completed(self._handshake_event.wait(), *(asyncio.shield(t) for t in lifetime_tasks))
+
+        return lifetime_tasks
+
+    async def _keep_alive(self) -> None:
+        assert self._handshake_event is not None
+
+        lifetime_tasks: typing.Tuple[asyncio.Task[None], ...] = ()
+        last_started_at = -float("inf")
+        backoff = rate_limits.ExponentialBackOff(base=_BACKOFF_BASE, maximum=_BACKOFF_CAP)
+
+        while True:
+            self._handshake_event.clear()
+
+            if time.monotonic() - last_started_at < _BACKOFF_WINDOW:
+                backoff_time = next(backoff)
+                self._logger.info("backing off reconnecting for %.2fs", backoff_time)
+                await asyncio.sleep(backoff_time)
 
             try:
-                if self._seq is not None:
-                    self._logger.debug("resuming session %s", self._session_id)
-                    await self._resume()
-                else:
-                    self._logger.debug("identifying with new session")
-                    await self._identify()
+                last_started_at = time.monotonic()
+                lifetime_tasks = await self._connect()
 
-                if self._closing_event.is_set():
-                    self._logger.debug(
-                        "closing flag was set during handshake, disconnecting with GOING AWAY "
-                        "(_run_once => do not reconnect)"
-                    )
-                    await self._get_ws().send_close(
-                        code=errors.ShardCloseCode.GOING_AWAY, message=b"shard disconnecting"
-                    )
-                    return False
-
-                # Event polling.
-                while not self._closing_event.is_set() and not heartbeat_task.done() and not heartbeat_task.cancelled():
-                    try:
-                        result = await self._poll_events()
-
-                        if result is not None:
-                            return result
-                    except asyncio.TimeoutError:
-                        # We should check if the shard is still alive and then poll again after.
-                        pass
-
-                # If the heartbeat died due to an error, it should be raised here.
-                # This will currently allow us to try to resume if that happens
-                # We return True if zombied.
-                if await heartbeat_task:
-                    now = time.monotonic()
-                    self._logger.error(
-                        "connection is a zombie, last heartbeat sent %.2fs ago",
-                        now - self._last_heartbeat_sent,
-                    )
-                    self._logger.debug("will attempt to reconnect (_run_once => reconnect)")
-                    return True
+                if not self._handshake_event.is_set():
+                    continue
 
-                self._logger.debug(
-                    "shard has requested graceful termination, so will not attempt to reconnect "
-                    "(_run_once => do not reconnect)"
-                )
-                await self._get_ws().send_close(
-                    code=errors.ShardCloseCode.GOING_AWAY,
-                    message=b"shard disconnecting",
-                )
-                return False
+                await aio.first_completed(*lifetime_tasks)
 
-            finally:
-                heartbeat_task.cancel()
+                # Since nothing went wrong, we can reset the backoff and try again
+                backoff.reset()
 
-        except Exception as ex:
-            exception = ex
-            raise
+            except errors.GatewayConnectionError as ex:
+                self._logger.warning("failed to communicate with server, reason was: %r. Will retry shortly", ex.reason)
 
-        finally:
-            ws = self._ws
-            self._ws = None
-            await exit_stack.aclose()
-            if dispatch_disconnect:
-                # If we managed to connect, we must always send the DISCONNECT event
-                # afterwards.
-                self._event_manager.dispatch(self._event_factory.deserialize_disconnected_event(self))
+            except errors.GatewayServerClosedConnectionError as ex:
+                if not ex.can_reconnect:
+                    self._logger.info(
+                        "server has closed the connection permanently [code:%s, reason:%]",
+                        ex.code,
+                        ex.reason,
+                    )
+                    raise
+
+                self._logger.info(
+                    "server has closed the connection, will attempt to reconnect [code:%s, reason:%s]",
+                    ex.code,
+                    ex.reason,
+                )
 
-            # Ignore errors if we are closing
-            if self._closing_event.is_set() or self._closed_event.is_set():
-                return False
+                # We don't want to back off from this. If Discord keep closing the connection, it is their issue.
+                # If we back off here, we'll find a mass outage will prevent shards from becoming healthy on
+                # reconnect in large sharded bots for a very long period of time.
+                backoff.reset()
 
-            # Check if we made the socket close or handled it. If we didn't, we should always try to
-            # reconnect, as aiohttp is probably closing it internally without telling us properly.
-            if exception is None and not ws.sent_close:  # type: ignore[union-attr]
-                return True
+            except errors.GatewayError as ex:
+                self._logger.error("encountered generic gateway error", exc_info=ex)
+                raise
 
-    async def _send_heartbeat(self) -> None:
-        await self._send_json({_OP: _HEARTBEAT, _D: self._seq})
-        self._last_heartbeat_sent = time.monotonic()
+            except asyncio.CancelledError:
+                self._is_closing = True
+                return
 
-    @staticmethod
-    def _serialize_activity(activity: typing.Optional[presences.Activity]) -> data_binding.JSONish:
-        if activity is None:
-            return None
+            except Exception as ex:
+                self._logger.error("encountered some unhandled error", exc_info=ex)
+                raise
 
-        return {"name": activity.name, "type": int(activity.type), "url": activity.url}
+            finally:
+                # Cancel any left-over tasks
+                for task in lifetime_tasks:
+                    if not task.done() and not task.cancelled():
+                        task.cancel()
+
+                        try:
+                            await task
+                        except asyncio.CancelledError:
+                            pass
+
+                # Close the ws
+                if self._ws:
+                    ws = self._ws
+                    self._ws = None
+
+                    if self._is_closing:
+                        await ws.send_close(
+                            code=errors.ShardCloseCode.GOING_AWAY, message=b"shard disconnecting permanently"
+                        )
+                    else:
+                        await ws.send_close(code=_RESUME_CLOSE_CODE, message=b"shard disconnecting temporarily")
+
+                    self._event_manager.dispatch(self._event_factory.deserialize_disconnected_event(self))
 
     def _serialize_and_store_presence_payload(
         self,
@@ -1014,8 +951,6 @@ def _serialize_and_store_presence_payload(
         status: undefined.UndefinedOr[presences.Status] = undefined.UNDEFINED,
         activity: undefined.UndefinedNoneOr[presences.Activity] = undefined.UNDEFINED,
     ) -> data_binding.JSONObject:
-        payload = data_binding.JSONObjectBuilder()
-
         if activity is undefined.UNDEFINED:
             activity = self._activity
         else:
@@ -1036,50 +971,11 @@ def _serialize_and_store_presence_payload(
         else:
             self._is_afk = afk
 
-        payload.put("since", idle_since, conversion=self._serialize_datetime)
+        payload = data_binding.JSONObjectBuilder()
+        payload.put("since", idle_since, conversion=_serialize_datetime)
         payload.put("afk", afk)
-        payload.put("game", activity, conversion=self._serialize_activity)
+        payload.put("game", activity, conversion=_serialize_activity)
         # Sending "offline" to the gateway won't do anything, we will have to
         # send "invisible" instead for this to work.
-        if status is presences.Status.OFFLINE:
-            payload.put("status", "invisible")
-        else:
-            payload.put("status", status)
+        payload.put("status", "invisible" if status is presences.Status.OFFLINE else status)
         return payload
-
-    @staticmethod
-    def _serialize_datetime(dt: typing.Optional[datetime.datetime]) -> typing.Optional[int]:
-        if dt is None:
-            return None
-
-        return int(dt.timestamp() * 1_000)
-
-    async def _wait_for_hello(self) -> asyncio.Task[bool]:
-        # Expect HELLO.
-        payload = await self._get_ws().receive_json()
-        if payload[_OP] != _HELLO:
-            self._logger.debug(
-                "expected HELLO opcode, received %s which makes no sense, closing with PROTOCOL ERROR ",
-                "(_run_once => raise and do not reconnect)",
-                payload[_OP],
-            )
-            await self._get_ws().send_close(
-                code=errors.ShardCloseCode.PROTOCOL_ERROR,
-                message=b"Expected HELLO op",
-            )
-            raise errors.GatewayError(f"Expected opcode {_HELLO}, but received {payload[_OP]}")
-
-        if self._closing_event.is_set():
-            self._logger.debug(
-                "closing flag was set before we could handshake, disconnecting with GOING AWAY "
-                "(_run_once => do not reconnect)"
-            )
-            await self._get_ws().send_close(
-                code=errors.ShardCloseCode.GOING_AWAY,
-                message=b"shard disconnecting",
-            )
-            raise asyncio.CancelledError("closing flag was set before we could handshake")
-
-        heartbeat_interval = float(payload[_D]["heartbeat_interval"]) / 1_000.0
-        heartbeat_task = asyncio.create_task(self._heartbeat(heartbeat_interval))
-        return heartbeat_task
diff --git a/hikari/internal/aio.py b/hikari/internal/aio.py
index c7042609..b7dae559 100644
--- a/hikari/internal/aio.py
+++ b/hikari/internal/aio.py
@@ -31,13 +31,17 @@
     "is_async_iterable",
     "first_completed",
     "all_of",
+    "destroy_loop",
 )
 
 import asyncio
 import inspect
+import sys
 import typing
 
 if typing.TYPE_CHECKING:
+    import logging
+
     # typing_extensions is a dependency of mypy, and pyright vendors it.
     import typing_extensions
 
@@ -126,8 +130,9 @@ async def first_completed(
         If more than one awaitable is completed before entering this call, then
         the first future is always returned.
     """
-    fs = list(map(asyncio.ensure_future, aws))
+    fs = tuple(map(asyncio.ensure_future, aws))
     iterator = asyncio.as_completed(fs, timeout=timeout)
+
     try:
         await next(iterator)
     except asyncio.CancelledError:
@@ -164,10 +169,9 @@ async def all_of(
     Returns
     -------
     typing.Sequence[T_co]
-        The results of each awaitable in the order they were provided invoked
-        in.
+        The results of each awaitable in the order they were invoked in.
     """
-    fs = list(map(asyncio.ensure_future, aws))
+    fs = tuple(map(asyncio.ensure_future, aws))
     gatherer = asyncio.gather(*fs)
 
     try:
@@ -188,9 +192,7 @@ async def all_of(
 
         gatherer.cancel()
         try:
-            # coverage.py will complain that this is not fully covered, as the
-            # "except" block will always be hit. This is intentional.
-            await gatherer  # pragma: no cover
+            await gatherer
         except asyncio.CancelledError:
             pass
 
@@ -217,3 +219,66 @@ def get_or_make_loop() -> asyncio.AbstractEventLoop:
     loop = asyncio.new_event_loop()
     asyncio.set_event_loop(loop)
     return loop
+
+
+async def _gather(coros: typing.Iterator[typing.Awaitable[typing.Any]]) -> None:
+    # Calling asyncio.gather outside of a running event loop isn't safe and
+    # will lead to RuntimeErrors in later versions of python, so this call is
+    # kept within a coroutine function.
+    await asyncio.gather(*coros)
+
+
+def destroy_loop(loop: asyncio.AbstractEventLoop, logger: logging.Logger) -> None:
+    """Destroy the passed loop.
+
+    Parameters
+    ----------
+    loop : asyncio.AbstractEventLoop
+        The loop to destroy
+    logger : logging.Logger
+        The logger to use for logging
+    """
+
+    async def murder(future: asyncio.Future[typing.Any]) -> None:
+        # These include _GatheringFuture which must be awaited if the children
+        # throw an asyncio.CancelledError, otherwise it will spam logs with warnings
+        # about exceptions not being retrieved before GC.
+        try:
+            logger.debug("killing %s", future)
+            future.cancel()
+            await future
+        except asyncio.CancelledError:
+            pass
+        except Exception as ex:
+            loop.call_exception_handler(
+                {
+                    "message": "Future raised unexpected exception after requesting cancellation",
+                    "exception": ex,
+                    "future": future,
+                }
+            )
+
+    remaining_tasks = tuple(t for t in asyncio.all_tasks(loop) if not t.done())
+
+    if remaining_tasks:
+        logger.debug("terminating %s remaining tasks forcefully", len(remaining_tasks))
+        loop.run_until_complete(_gather((murder(task) for task in remaining_tasks)))
+    else:
+        logger.debug("No remaining tasks exist, good job!")
+
+    if sys.version_info >= (3, 9):
+        logger.debug("shutting down default executor")
+        try:
+            # This seems to raise a NotImplementedError when running with uvloop.
+            loop.run_until_complete(loop.shutdown_default_executor())
+        except NotImplementedError:
+            pass
+
+    logger.debug("shutting down asyncgens")
+    loop.run_until_complete(loop.shutdown_asyncgens())
+
+    logger.debug("closing event loop")
+    loop.close()
+
+    # Closed loops cannot be re-used so it should also be un-set.
+    asyncio.set_event_loop(None)
diff --git a/hikari/internal/net.py b/hikari/internal/net.py
index 4e0f0cd8..3aa1ffc8 100644
--- a/hikari/internal/net.py
+++ b/hikari/internal/net.py
@@ -119,7 +119,6 @@ def create_client_session(
     http_settings: config.HTTPSettings,
     raise_for_status: bool,
     trust_env: bool,
-    ws_response_cls: typing.Type[aiohttp.ClientWebSocketResponse] = aiohttp.ClientWebSocketResponse,
 ) -> aiohttp.ClientSession:
     """Generate a client session using the given settings.
 
@@ -138,7 +137,7 @@ def create_client_session(
         The connector to use.
     connector_owner : builtins.bool
         If `builtins.True`, then the client session will close the
-        connector on shutdown. Otherwise, you must do it manually.
+        connector on shut down. Otherwise, you must do it manually.
     http_settings : hikari.config.HTTPSettings
         HTTP settings to use.
     raise_for_status : builtins.bool
@@ -147,10 +146,6 @@ def create_client_session(
     trust_env : builtins.bool
         `builtins.True` to trust anything in environment variables
         and the `netrc` file, `builtins.False` to ignore it.
-    ws_response_cls : typing.Type[aiohttp.ClientWebSocketResponse]
-        The websocket response class to use.
-
-        Defaults to `aiohttp.ClientWebSocketResponse`.
 
     Returns
     -------
@@ -169,5 +164,4 @@ def create_client_session(
         ),
         trust_env=trust_env,
         version=aiohttp.HttpVersion11,
-        ws_response_class=ws_response_cls,
     )
diff --git a/hikari/internal/signals.py b/hikari/internal/signals.py
new file mode 100644
index 00000000..0be4f27d
--- /dev/null
+++ b/hikari/internal/signals.py
@@ -0,0 +1,129 @@
+# -*- coding: utf-8 -*-
+# cython: language_level=3
+# Copyright (c) 2020 Nekokatt
+# Copyright (c) 2021-present davfsa
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+"""Signal handling utilities."""
+
+from __future__ import annotations
+
+__all__: typing.Sequence[str] = ("handle_interrupts",)
+
+import asyncio
+import contextlib
+import logging
+import signal
+import threading
+import traceback
+import types
+import typing
+
+from hikari import errors
+from hikari.internal import ux
+
+_INTERRUPT_SIGNALS: typing.Tuple[str, ...] = ("SIGINT", "SIGTERM")
+_LOGGER: typing.Final[logging.Logger] = logging.getLogger("hikari.signals")
+
+
+def _raise_interrupt(signum: int) -> typing.NoReturn:
+    signame = signal.strsignal(signum)
+    assert signame is not None
+
+    raise errors.HikariInterrupt(signum, signame)
+
+
+def _interrupt_handler(
+    loop: asyncio.AbstractEventLoop,
+) -> typing.Callable[[int, typing.Optional[types.FrameType]], None]:
+    loop_thread_id = threading.get_native_id()
+
+    def handler(signum: int, frame: typing.Optional[types.FrameType]) -> None:
+        # The loop may or may not be running, depending on the state of the application when this occurs.
+        # Signals on POSIX only occur on the main thread usually, too, so we need to ensure this is
+        # threadsafe.
+        # We log native thread IDs purely for debugging purposes.
+        if _LOGGER.isEnabledFor(ux.TRACE):
+            _LOGGER.log(
+                ux.TRACE,
+                "interrupt %s occurred on thread %s, process on thread %s will be notified shortly\n"
+                "Stacktrace for developer sanity:\n%s",
+                signum,
+                threading.get_native_id(),
+                loop_thread_id,
+                "".join(traceback.format_stack(frame)),
+            )
+
+        loop.call_soon_threadsafe(_raise_interrupt, signum)
+
+    return handler
+
+
+@contextlib.contextmanager
+def handle_interrupts(
+    enabled: typing.Optional[bool],
+    loop: asyncio.AbstractEventLoop,
+    propagate_interrupts: bool,
+) -> typing.Generator[None, None, None]:
+    """Context manager which cleanly exits on signal interrupts.
+
+    Parameters
+    ----------
+    enabled : typing.Optional[bool]
+        Whether to enable the signal interrupts.
+
+        If set to `None`, then it will be enabled or not based on whether the running
+        thread is the main one or not.
+    loop : asyncio.AbstractEventLoop
+        The event loop the interrupt will be raised in.
+    propagate_interrupts : bool
+        Whether to propagate interrupts.
+    """
+    if enabled is None:
+        enabled = threading.current_thread() is threading.main_thread()
+
+    if not enabled:
+        # NOOP context manager
+        yield
+        return
+
+    interrupt_handler = _interrupt_handler(loop)
+
+    for sig in _INTERRUPT_SIGNALS:
+        try:
+            signum = getattr(signal, sig)
+            signal.signal(signum, interrupt_handler)
+        except AttributeError:
+            _LOGGER.log(ux.TRACE, "signal %s is not implemented on your platform; skipping", sig)
+
+    try:
+        yield
+
+    except errors.HikariInterrupt:
+        if propagate_interrupts:
+            raise
+
+    finally:
+        for sig in _INTERRUPT_SIGNALS:
+            try:
+                signum = getattr(signal, sig)
+                signal.signal(signum, signal.SIG_DFL)
+            except AttributeError:
+                # Signal not implemented. We already logged this earlier.
+                pass
diff --git a/hikari/traits.py b/hikari/traits.py
index 0778dc6b..18040dae 100644
--- a/hikari/traits.py
+++ b/hikari/traits.py
@@ -560,133 +560,6 @@ class GatewayBotAware(
 
     __slots__: typing.Sequence[str] = ()
 
-    async def join(self, until_close: bool = True) -> None:
-        """Wait indefinitely until the application closes.
-
-        This can be placed in a task and cancelled without affecting the
-        application runtime itself. Any exceptions raised by shards will be
-        propagated to here.
-
-        Other Parameters
-        ----------------
-        until_close : builtins.bool
-            Defaults to `builtins.True`. If set, the waiter will stop as soon as
-            a request for shut down is processed. This can allow you to break
-            and begin closing your own resources.
-
-            If `builtins.False`, then this will wait until all shards' tasks
-            have died.
-        """
-        raise NotImplementedError
-
-    def run(
-        self,
-        *,
-        activity: typing.Optional[presences.Activity] = None,
-        afk: bool = False,
-        close_passed_executor: bool = False,
-        idle_since: typing.Optional[datetime.datetime] = None,
-        ignore_session_start_limit: bool = False,
-        large_threshold: int = 250,
-        status: presences.Status = presences.Status.ONLINE,
-        shard_ids: typing.Optional[typing.AbstractSet[int]] = None,
-        shard_count: typing.Optional[int] = None,
-    ) -> None:
-        """Start the bot and block until it's finished running.
-
-        Other Parameters
-        ----------------
-        activity : typing.Optional[hikari.presences.Activity]
-            The initial activity to display in the bot user presence, or
-            `builtins.None` (default) to not show any.
-        afk : builtins.bool
-            The initial AFK state to display in the bot user presence, or
-            `builtins.False` (default) to not show any.
-        close_executor : builtins.bool
-            Defaults to `builtins.False`. If `builtins.True`, any custom
-            `concurrent.futures.Executor` passed to the constructor will be
-            shut down when the application terminates. This does not affect the
-            default executor associated with the event loop, and will not
-            do anything if you do not provide a custom executor to the
-            constructor.
-        idle_since : typing.Optional[datetime.datetime]
-            The `datetime.datetime` the user should be marked as being idle
-            since, or `builtins.None` (default) to not show this.
-        ignore_session_start_limit : builtins.bool
-            Defaults to `builtins.False`. If `builtins.False`, then attempting
-            to start more sessions than you are allowed in a 24 hour window
-            will throw a `hikari.errors.GatewayError` rather than going ahead
-            and hitting the IDENTIFY limit, which may result in your token
-            being reset. Setting to `builtins.True` disables this behavior.
-        large_threshold : builtins.int
-            Threshold for members in a guild before it is treated as being
-            "large" and no longer sending member details in the `GUILD CREATE`
-            event. Defaults to `250`.
-        shard_ids : typing.Optional[typing.AbstractSet[builtins.int]]
-            The shard IDs to create shards for. If not `builtins.None`, then
-            a non-`None` `shard_count` must ALSO be provided. Defaults to
-            `builtins.None`, which means the Discord-recommended count is used
-            for your application instead.
-        shard_count : typing.Optional[builtins.int]
-            The number of shards to use in the entire distributed application.
-            Defaults to `builtins.None` which results in the count being
-            determined dynamically on startup.
-        status : hikari.presences.Status
-            The initial status to show for the user presence on startup.
-            Defaults to `hikari.presences.Status.ONLINE`.
-        """
-        raise NotImplementedError
-
-    async def start(
-        self,
-        *,
-        activity: typing.Optional[presences.Activity] = None,
-        afk: bool = False,
-        idle_since: typing.Optional[datetime.datetime] = None,
-        ignore_session_start_limit: bool = False,
-        large_threshold: int = 250,
-        shard_ids: typing.Optional[typing.AbstractSet[int]] = None,
-        shard_count: typing.Optional[int] = None,
-        status: presences.Status = presences.Status.ONLINE,
-    ) -> None:
-        """Start the bot, wait for all shards to become ready, and then return.
-
-        Other Parameters
-        ----------------
-        activity : typing.Optional[hikari.presences.Activity]
-            The initial activity to display in the bot user presence, or
-            `builtins.None` (default) to not show any.
-        afk : builtins.bool
-            The initial AFK state to display in the bot user presence, or
-            `builtins.False` (default) to not show any.
-        idle_since : typing.Optional[datetime.datetime]
-            The `datetime.datetime` the user should be marked as being idle
-            since, or `builtins.None` (default) to not show this.
-        ignore_session_start_limit : builtins.bool
-            Defaults to `builtins.False`. If `builtins.False`, then attempting
-            to start more sessions than you are allowed in a 24 hour window
-            will throw a `hikari.errors.GatewayError` rather than going ahead
-            and hitting the IDENTIFY limit, which may result in your token
-            being reset. Setting to `builtins.True` disables this behavior.
-        large_threshold : builtins.int
-            Threshold for members in a guild before it is treated as being
-            "large" and no longer sending member details in the `GUILD CREATE`
-            event. Defaults to `250`.
-        shard_ids : typing.Optional[typing.AbstractSet[builtins.int]]
-            The shard IDs to create shards for. If not `builtins.None`, then
-            a non-`None` `shard_count` must ALSO be provided. Defaults to
-            `builtins.None`, which means the Discord-recommended count is used
-            for your application instead.
-        shard_count : typing.Optional[builtins.int]
-            The number of shards to use in the entire distributed application.
-            Defaults to `builtins.None` which results in the count being
-            determined dynamically on startup.
-        status : hikari.presences.Status
-            The initial status to show for the user presence on startup.
-            Defaults to `hikari.presences.Status.ONLINE`.
-        """
-        raise NotImplementedError
-
 
 @typing.runtime_checkable
 class RESTBotAware(InteractionServerAware, Runnable, fast_protocol.FastProtocolChecking, typing.Protocol):


Test patch:
diff --git a/tests/hikari/hikari_test_helpers.py b/tests/hikari/hikari_test_helpers.py
index 18b54375..d46c2b5c 100644
--- a/tests/hikari/hikari_test_helpers.py
+++ b/tests/hikari/hikari_test_helpers.py
@@ -156,6 +156,22 @@ def ensure_occurs_quickly():
         yield
 
 
+class ContextManagerMock:
+    enter_count = 0
+    exit_count = 0
+
+    def __enter__(self):
+        self.enter_count += 1
+        return self
+
+    def __exit__(self, *args):
+        self.exit_count += 1
+
+    def assert_used_once(self):
+        assert self.enter_count == 1
+        assert self.exit_count == 1
+
+
 class AsyncContextManagerMock:
     aenter_count = 0
     aexit_count = 0
diff --git a/tests/hikari/impl/test_bot.py b/tests/hikari/impl/test_bot.py
index 0a632bc2..a1b0a208 100644
--- a/tests/hikari/impl/test_bot.py
+++ b/tests/hikari/impl/test_bot.py
@@ -21,7 +21,6 @@
 # SOFTWARE.
 import asyncio
 import contextlib
-import signal
 import sys
 import warnings
 
@@ -43,7 +42,9 @@
 from hikari.impl import shard as shard_impl
 from hikari.impl import voice as voice_impl
 from hikari.internal import aio
+from hikari.internal import signals
 from hikari.internal import ux
+from tests.hikari import hikari_test_helpers
 
 
 @pytest.mark.parametrize("activity", [undefined.UNDEFINED, None])
@@ -332,55 +333,43 @@ def test_voice(self, bot, voice):
     def test_rest(self, bot, rest):
         assert bot.rest is rest
 
-    def test_is_alive(self, bot):
-        bot._is_alive = True
+    @pytest.mark.parametrize(("closed_event", "expected"), [("something", True), (None, False)])
+    def test_is_alive(self, bot, closed_event, expected):
+        bot._closed_event = closed_event
 
-        assert bot.is_alive is True
-
-    def test_check_if_alive_when_False(self, bot):
-        bot._is_alive = False
-
-        with pytest.raises(errors.ComponentStateConflictError):
-            bot._check_if_alive()
+        assert bot.is_alive is expected
 
     def test_check_if_alive(self, bot):
-        bot._is_alive = True
+        bot._closed_event = object()
 
         bot._check_if_alive()
 
-    @pytest.mark.asyncio()
-    async def test_close(self, bot):
-        with mock.patch.object(bot_impl.GatewayBot, "_close") as internal_close:
-            with mock.patch.object(bot_impl.GatewayBot, "_check_if_alive") as check_if_alive:
-                await bot.close()
+    def test_check_if_alive_when_False(self, bot):
+        bot._closed_event = None
 
-        check_if_alive.assert_called_once_with()
-        internal_close.assert_awaited_once_with()
+        with pytest.raises(errors.ComponentStateConflictError):
+            bot._check_if_alive()
 
     @pytest.mark.asyncio()
-    async def test__close_when_already_closed(self, bot):
-        bot._closing_event = None
-        bot._closed_event = None
-        bot._is_alive = True
-
-        await bot._close()
+    async def test_close_when_already_closed(self, bot):
+        bot._closed_event = mock.Mock()
 
-        assert bot._is_alive is True
+        with pytest.raises(errors.ComponentStateConflictError):
+            await bot.close()
 
     @pytest.mark.asyncio()
-    async def test__close_when_already_closing(self, bot):
-        bot._closing_event = object()
-        bot._closed_event = mock.AsyncMock()
-        bot._is_alive = True
+    async def test_close_when_already_closing(self, bot):
+        bot._closed_event = mock.Mock()
+        bot._closing_event = mock.Mock(is_set=mock.Mock(return_value=True))
 
-        await bot._close()
+        with mock.patch.object(bot_impl.GatewayBot, "join") as join:
+            await bot.close()
 
-        bot._closed_event.wait.assert_awaited_once_with()
-        assert bot._is_alive is True
+        join.assert_awaited_once_with()
+        bot._closed_event.set.assert_not_called()
 
     @pytest.mark.asyncio()
-    @pytest.mark.parametrize("is_alive", [True, False])
-    async def test__close(self, bot, event_manager, event_factory, rest, voice, cache, is_alive):
+    async def test_close(self, bot, event_manager, event_factory, rest, voice, cache):
         def null_call(arg):
             return arg
 
@@ -408,16 +397,14 @@ def assert_awaited_once(self):
         stack.enter_context(mock.patch.object(asyncio, "as_completed", side_effect=null_call))
         ensure_future = stack.enter_context(mock.patch.object(asyncio, "ensure_future", side_effect=null_call))
         get_running_loop = stack.enter_context(mock.patch.object(asyncio, "get_running_loop"))
-        new_event = stack.enter_context(mock.patch.object(asyncio, "Event"))
         mock_future = mock.Mock()
         get_running_loop.return_value.create_future.return_value = mock_future
 
         event_manager.dispatch = mock.AsyncMock()
         rest.close = AwaitableMock()
         voice.close = AwaitableMock()
+        bot._closed_event = close_event = mock.Mock()
         bot._closing_event = closing_event = mock.Mock(is_set=mock.Mock(return_value=False))
-        bot._closed_event = None
-        bot._is_alive = is_alive
         error = RuntimeError()
         shard0 = mock.Mock(id=0, close=AwaitableMock())
         shard1 = mock.Mock(id=1, close=AwaitableMock(error))
@@ -425,23 +412,23 @@ def assert_awaited_once(self):
         bot._shards = {0: shard0, 1: shard1, 2: shard2}
 
         with stack:
-            await bot._close()
+            await bot.close()
 
         # Events and args
+        close_event.set.assert_called_once_with()
+        assert bot._closed_event is None
+
         closing_event.set.assert_called_once_with()
         assert bot._closing_event is None
-        new_event.return_value.set.assert_called_once_with()
-        assert bot._closed_event is None
-        assert bot._is_alive is False
 
         # Closing components
         ensure_future.assert_has_calls(
             [
-                mock.call(rest.close()),
-                mock.call(voice.close()),
                 mock.call(shard0.close()),
                 mock.call(shard1.close()),
                 mock.call(shard2.close()),
+                mock.call(voice.close()),
+                mock.call(rest.close()),
             ],
         )
 
@@ -455,7 +442,7 @@ def assert_awaited_once(self):
         get_running_loop.assert_called_once_with()
         get_running_loop.return_value.call_exception_handler.assert_called_once_with(
             {
-                "message": "shard 1 raised an exception during shutdown",
+                "message": "shard 1 raised an exception during shut down",
                 "future": shard1.close(),
                 "exception": error,
             }
@@ -465,16 +452,12 @@ def assert_awaited_once(self):
         assert bot._shards == {}
         cache.clear.assert_called_once_with()
 
-        if is_alive:
-            # Dispatching events in the right order
-            event_manager.dispatch.assert_has_calls(
-                [
-                    mock.call(event_factory.deserialize_stopping_event.return_value),
-                    mock.call(event_factory.deserialize_stopped_event.return_value),
-                ],
-            )
-        else:
-            event_manager.dispatch.assert_not_called()
+        event_manager.dispatch.assert_has_calls(
+            [
+                mock.call(event_factory.deserialize_stopping_event.return_value),
+                mock.call(event_factory.deserialize_stopped_event.return_value),
+            ],
+        )
 
     def test_dispatch(self, bot, event_manager):
         event = object()
@@ -491,42 +474,19 @@ def test_get_listeners(self, bot, event_manager):
         event_manager.get_listeners.assert_called_once_with(event, polymorphic=False)
 
     @pytest.mark.asyncio()
-    async def test_join_when_not_until_close(self, bot, event_manager):
-        shard0 = mock.Mock()
-        shard1 = mock.Mock()
-        shard2 = mock.Mock()
-        bot._shards = {0: shard0, 1: shard1, 2: shard2}
+    async def test_join(self, bot, event_manager):
+        bot._closed_event = mock.AsyncMock()
 
-        with mock.patch.object(aio, "first_completed") as first_completed:
-            with mock.patch.object(bot_impl.GatewayBot, "_check_if_alive") as check_if_alive:
-                await bot.join(until_close=False)
+        await bot.join()
 
-        check_if_alive.assert_called_once_with()
-        first_completed.assert_awaited_once_with(
-            *(shard0.join.return_value, shard1.join.return_value, shard2.join.return_value)
-        )
+        bot._closed_event.wait.assert_awaited_once_with()
 
     @pytest.mark.asyncio()
-    async def test_join_when_until_close(self, bot):
-        shard0 = mock.Mock()
-        shard1 = mock.Mock()
-        shard2 = mock.Mock()
-        bot._shards = {0: shard0, 1: shard1, 2: shard2}
-        bot._closing_event = mock.Mock()
-
-        with mock.patch.object(aio, "first_completed") as first_completed:
-            with mock.patch.object(bot_impl.GatewayBot, "_check_if_alive") as check_if_alive:
-                await bot.join(until_close=True)
+    async def test_join_when_not_running(self, bot, event_manager):
+        bot._closed_event = None
 
-        check_if_alive.assert_called_once_with()
-        first_completed.assert_awaited_once_with(
-            *(
-                shard0.join.return_value,
-                shard1.join.return_value,
-                shard2.join.return_value,
-                bot._closing_event.wait(),
-            )
-        )
+        with pytest.raises(errors.ComponentStateConflictError):
+            await bot.join()
 
     def test_listen(self, bot, event_manager):
         event = object()
@@ -542,7 +502,7 @@ def test_print_banner(self, bot):
         print_banner.assert_called_once_with("testing", False, True, extra_args={"test_key": "test_value"})
 
     def test_run_when_already_running(self, bot):
-        bot._is_alive = True
+        bot._closed_event = object()
 
         with pytest.raises(errors.ComponentStateConflictError):
             bot.run()
@@ -555,7 +515,9 @@ def test_run_with_asyncio_debug(self, bot):
         stack = contextlib.ExitStack()
         stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "start", new=mock.Mock()))
         stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "join", new=mock.Mock()))
-        stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "_close", new=mock.Mock()))
+        stack.enter_context(
+            mock.patch.object(signals, "handle_interrupts", return_value=hikari_test_helpers.ContextManagerMock())
+        )
         loop = stack.enter_context(mock.patch.object(aio, "get_or_make_loop")).return_value
 
         with stack:
@@ -567,7 +529,9 @@ def test_run_with_coroutine_tracking_depth(self, bot):
         stack = contextlib.ExitStack()
         stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "start", new=mock.Mock()))
         stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "join", new=mock.Mock()))
-        stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "_close", new=mock.Mock()))
+        stack.enter_context(
+            mock.patch.object(signals, "handle_interrupts", return_value=hikari_test_helpers.ContextManagerMock())
+        )
         stack.enter_context(mock.patch.object(aio, "get_or_make_loop"))
         coroutine_tracking_depth = stack.enter_context(
             mock.patch.object(sys, "set_coroutine_origin_tracking_depth", create=True, side_effect=AttributeError)
@@ -578,54 +542,13 @@ def test_run_with_coroutine_tracking_depth(self, bot):
 
         coroutine_tracking_depth.assert_called_once_with(100)
 
-    def test_run_with_enable_signal_handlers(self, bot):
-        stack = contextlib.ExitStack()
-        stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "start", new=mock.Mock()))
-        stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "join", new=mock.Mock()))
-        stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "_close", new=mock.Mock()))
-        stack.enter_context(mock.patch.object(aio, "get_or_make_loop"))
-        signal_function = stack.enter_context(
-            mock.patch.object(signal, "signal", side_effect=[None, AttributeError, None, AttributeError])
-        )
-
-        with stack:
-            bot.run(close_loop=False, enable_signal_handlers=True)
-
-        # We have these twice because they will also be called on cleanup too
-        expected_signals = [
-            signal.Signals.SIGINT,
-            signal.Signals.SIGTERM,
-            signal.Signals.SIGINT,
-            signal.Signals.SIGTERM,
-        ]
-        assert [signal_function.call_args_list[i][0][0] for i in range(signal_function.call_count)] == expected_signals
-
-    @pytest.mark.parametrize("logging", [True, False])
-    def test_run_with_propagate_interrupts(self, bot, logging):
-        def raise_signal(*args, **kwargs):
-            signal.raise_signal(signal.Signals.SIGTERM)
-
-        stack = contextlib.ExitStack()
-        stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "start", new=mock.Mock()))
-        stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "join", new=raise_signal))
-        stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "_close", new=mock.Mock()))
-        stack.enter_context(mock.patch.object(bot_impl, "_LOGGER", isEnabledFor=mock.Mock(return_value=logging)))
-        loop = stack.enter_context(mock.patch.object(aio, "get_or_make_loop")).return_value
-        set_close_flag = stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "_set_close_flag", new=mock.Mock()))
-        run_coroutine_threadsafe = stack.enter_context(mock.patch.object(asyncio, "run_coroutine_threadsafe"))
-
-        with stack:
-            with pytest.raises(errors.HikariInterrupt, match=rf"(15, '{signal.strsignal(signal.Signals.SIGTERM)}')"):
-                bot.run(close_loop=False, propagate_interrupts=True)
-
-        run_coroutine_threadsafe.assert_called_once_with(set_close_flag.return_value, loop)
-        set_close_flag.assert_called_once_with(signal.strsignal(signal.Signals.SIGTERM), 15)
-
     def test_run_with_close_passed_executor(self, bot):
         stack = contextlib.ExitStack()
         stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "start", new=mock.Mock()))
         stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "join", new=mock.Mock()))
-        stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "_close", new=mock.Mock()))
+        stack.enter_context(
+            mock.patch.object(signals, "handle_interrupts", return_value=hikari_test_helpers.ContextManagerMock())
+        )
         stack.enter_context(mock.patch.object(aio, "get_or_make_loop"))
         executor = mock.Mock()
         bot._executor = executor
@@ -636,18 +559,21 @@ def test_run_with_close_passed_executor(self, bot):
         executor.shutdown.assert_called_once_with(wait=True)
         assert bot._executor is None
 
-    def test_run_with_close_loop(self, bot):
+    def test_run_when_close_loop(self, bot):
         stack = contextlib.ExitStack()
+        logger = stack.enter_context(mock.patch.object(bot_impl, "_LOGGER"))
         stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "start", new=mock.Mock()))
         stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "join", new=mock.Mock()))
-        stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "_close", new=mock.Mock()))
-        destroy_loop = stack.enter_context(mock.patch.object(bot_impl, "_destroy_loop"))
+        stack.enter_context(
+            mock.patch.object(signals, "handle_interrupts", return_value=hikari_test_helpers.ContextManagerMock())
+        )
+        destroy_loop = stack.enter_context(mock.patch.object(aio, "destroy_loop"))
         loop = stack.enter_context(mock.patch.object(aio, "get_or_make_loop")).return_value
 
         with stack:
             bot.run(close_loop=True)
 
-        destroy_loop.assert_called_once_with(loop)
+        destroy_loop.assert_called_once_with(loop, logger)
 
     def test_run(self, bot):
         activity = object()
@@ -663,7 +589,9 @@ def test_run(self, bot):
         stack = contextlib.ExitStack()
         start_function = stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "start", new=mock.Mock()))
         join_function = stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "join", new=mock.Mock()))
-        close_function = stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "_close", new=mock.Mock()))
+        handle_interrupts = stack.enter_context(
+            mock.patch.object(signals, "handle_interrupts", return_value=hikari_test_helpers.ContextManagerMock())
+        )
         loop = stack.enter_context(mock.patch.object(aio, "get_or_make_loop")).return_value
 
         with stack:
@@ -689,7 +617,6 @@ def test_run(self, bot):
             [
                 mock.call(start_function.return_value),
                 mock.call(join_function.return_value),
-                mock.call(close_function.return_value),
             ]
         )
         start_function.assert_called_once_with(
@@ -703,22 +630,188 @@ def test_run(self, bot):
             shard_count=shard_count,
             status=status,
         )
+        handle_interrupts.assert_called_once_with(enabled=False, loop=loop, propagate_interrupts=False)
+        handle_interrupts.return_value.assert_used_once()
 
     @pytest.mark.asyncio()
     async def test_start_when_shard_ids_specified_without_shard_count(self, bot):
         with pytest.raises(TypeError, match=r"'shard_ids' must be passed with 'shard_count'"):
-            await bot.start(shard_ids={1})
+            await bot.start(shard_ids=(1,))
 
     @pytest.mark.asyncio()
     async def test_start_when_already_running(self, bot):
-        bot._is_alive = True
+        bot._closed_event = object()
 
         with pytest.raises(errors.ComponentStateConflictError):
             await bot.start()
 
-    @pytest.mark.skip("TODO")
-    def test_start(self, bot):
-        ...
+    @pytest.mark.asyncio()
+    async def test_start(self, bot, rest, voice, event_manager, event_factory, http_settings, proxy_settings):
+        class MockSessionStartLimit:
+            remaining = 10
+            reset_at = "now"
+            max_concurrency = 1
+
+        class MockInfo:
+            url = "yourmom.eu"
+            shard_count = 2
+            session_start_limit = MockSessionStartLimit()
+
+        shard1 = mock.Mock()
+        shard2 = mock.Mock()
+        shards_iter = iter((shard1, shard2))
+
+        mock_start_one_shard = mock.Mock()
+
+        def _mock_start_one_shard(*args, **kwargs):
+            bot._shards[kwargs["shard_id"]] = next(shards_iter)
+            return mock_start_one_shard(*args, **kwargs)
+
+        stack = contextlib.ExitStack()
+        stack.enter_context(mock.patch.object(bot_impl, "_validate_activity"))
+        stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "_start_one_shard", new=_mock_start_one_shard))
+        create_task = stack.enter_context(mock.patch.object(asyncio, "create_task"))
+        gather = stack.enter_context(mock.patch.object(asyncio, "gather"))
+        event = stack.enter_context(mock.patch.object(asyncio, "Event"))
+        first_completed = stack.enter_context(
+            mock.patch.object(aio, "first_completed", side_effect=[None, asyncio.TimeoutError, None])
+        )
+        check_for_updates = stack.enter_context(mock.patch.object(ux, "check_for_updates", new=mock.Mock()))
+
+        event_manager.dispatch = mock.AsyncMock()
+        rest.fetch_gateway_bot_info = mock.AsyncMock(return_value=MockInfo())
+
+        with stack:
+            await bot.start(
+                check_for_updates=True,
+                shard_ids=(2, 10),
+                shard_count=20,
+                activity="some activity",
+                afk=True,
+                idle_since="some idle since",
+                status="some status",
+                large_threshold=500,
+            )
+
+        check_for_updates.assert_called_once_with(http_settings, proxy_settings)
+        create_task.assert_called_once_with(check_for_updates.return_value, name="check for package updates")
+        rest.start.assert_called_once_with()
+        voice.start.assert_called_once_with()
+
+        assert event_manager.dispatch.call_count == 2
+        event_manager.dispatch.assert_has_awaits(
+            [
+                mock.call(event_factory.deserialize_starting_event.return_value),
+                mock.call(event_factory.deserialize_started_event.return_value),
+            ]
+        )
+
+        assert gather.call_count == 2
+        gather.assert_has_calls(
+            [mock.call(mock_start_one_shard.return_value), mock.call(mock_start_one_shard.return_value)]
+        )
+
+        assert mock_start_one_shard.call_count == 2
+        mock_start_one_shard.assert_has_calls(
+            [
+                mock.call(
+                    bot,
+                    activity="some activity",
+                    afk=True,
+                    idle_since="some idle since",
+                    status="some status",
+                    large_threshold=500,
+                    shard_id=i,
+                    shard_count=20,
+                    url="yourmom.eu",
+                )
+                for i in (2, 10)
+            ]
+        )
+
+        closing_event_closing_wait = event.return_value.wait.return_value
+        assert first_completed.await_count == 3
+
+        first_completed.assert_has_awaits(
+            [
+                # Shard 1
+                mock.call(closing_event_closing_wait, gather.return_value),
+                # Shard 2
+                mock.call(closing_event_closing_wait, shard1.join.return_value, timeout=5),
+                mock.call(closing_event_closing_wait, gather.return_value),
+            ]
+        )
+
+    @pytest.mark.asyncio()
+    async def test_start_when_request_close_mid_startup(self, bot, rest, voice, event_manager, event_factory):
+        class MockSessionStartLimit:
+            remaining = 10
+            reset_at = "now"
+            max_concurrency = 1
+
+        class MockInfo:
+            url = "yourmom.eu"
+            shard_count = 2
+            session_start_limit = MockSessionStartLimit()
+
+        # Assume that we already started one shard
+        shard1 = mock.Mock()
+        bot._shards = {"1": shard1}
+
+        stack = contextlib.ExitStack()
+        start_one_shard = stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "_start_one_shard"))
+        first_completed = stack.enter_context(mock.patch.object(aio, "first_completed"))
+        event = stack.enter_context(
+            mock.patch.object(asyncio, "Event", return_value=mock.Mock(is_set=mock.Mock(return_value=True)))
+        )
+
+        event_manager.dispatch = mock.AsyncMock()
+        rest.fetch_gateway_bot_info = mock.AsyncMock(return_value=MockInfo())
+
+        with stack:
+            await bot.start(shard_ids=(2, 10), shard_count=20, check_for_updates=False)
+
+        start_one_shard.assert_not_called()
+
+        first_completed.assert_called_once_with(
+            event.return_value.wait.return_value, shard1.join.return_value, timeout=5
+        )
+
+    @pytest.mark.asyncio()
+    async def test_start_when_shard_closed_mid_startup(self, bot, rest, voice, event_manager, event_factory):
+        class MockSessionStartLimit:
+            remaining = 10
+            reset_at = "now"
+            max_concurrency = 1
+
+        class MockInfo:
+            url = "yourmom.eu"
+            shard_count = 2
+            session_start_limit = MockSessionStartLimit()
+
+        # Assume that we already started one shard
+        shard1 = mock.Mock()
+        bot._shards = {"1": shard1}
+
+        stack = contextlib.ExitStack()
+        start_one_shard = stack.enter_context(mock.patch.object(bot_impl.GatewayBot, "_start_one_shard"))
+        first_completed = stack.enter_context(mock.patch.object(aio, "first_completed"))
+        event = stack.enter_context(
+            mock.patch.object(asyncio, "Event", return_value=mock.Mock(is_set=mock.Mock(return_value=False)))
+        )
+        stack.enter_context(pytest.raises(RuntimeError, match="One or more shards closed while starting"))
+
+        event_manager.dispatch = mock.AsyncMock()
+        rest.fetch_gateway_bot_info = mock.AsyncMock(return_value=MockInfo())
+
+        with stack:
+            await bot.start(shard_ids=(2, 10), shard_count=20, check_for_updates=False)
+
+        start_one_shard.assert_not_called()
+
+        first_completed.assert_called_once_with(
+            event.return_value.wait.return_value, shard1.join.return_value, timeout=5
+        )
 
     def test_stream(self, bot):
         event_type = object()
@@ -848,38 +941,24 @@ async def test_request_guild_members(self, bot):
             nonce="NONCE",
         )
 
-    @pytest.mark.asyncio()
-    async def test_set_close_flag(self, bot):
-        with mock.patch.object(bot_impl.GatewayBot, "_close") as close:
-            await bot._set_close_flag("Terminated", 15)
-
-        close.assert_awaited_once_with()
-
     @pytest.mark.asyncio()
     async def test_start_one_shard(self, bot):
         activity = object()
         status = object()
         bot._shards = {}
-        closing_event = mock.Mock()
-        shard = mock.Mock()
-        shard_obj = shard.return_value
-        shard_obj.is_alive = True
-
-        with mock.patch.object(aio, "first_completed", new=mock.AsyncMock()) as first_completed:
-            with mock.patch.object(shard_impl, "GatewayShardImpl", new=shard):
-                returned = await bot._start_one_shard(
-                    activity=activity,
-                    afk=True,
-                    idle_since=None,
-                    status=status,
-                    large_threshold=1000,
-                    shard_id=1,
-                    shard_count=3,
-                    url="https://some.website",
-                    closing_event=closing_event,
-                )
+        shard_obj = mock.Mock(is_alive=True, start=mock.AsyncMock())
 
-                assert returned is shard_obj
+        with mock.patch.object(shard_impl, "GatewayShardImpl", new=mock.Mock(return_value=shard_obj)) as shard:
+            await bot._start_one_shard(
+                activity=activity,
+                afk=True,
+                idle_since=None,
+                status=status,
+                large_threshold=1000,
+                shard_id=1,
+                shard_count=3,
+                url="https://some.website",
+            )
 
         shard.assert_called_once_with(
             http_settings=bot._http_settings,
@@ -897,29 +976,60 @@ async def test_start_one_shard(self, bot):
             token=bot._token,
             url="https://some.website",
         )
+        shard_obj.start.assert_awaited_once_with()
         assert bot._shards == {1: shard_obj}
-        first_completed.assert_awaited_once_with(shard_obj.start.return_value, closing_event.wait.return_value)
 
     @pytest.mark.asyncio()
     async def test_start_one_shard_when_not_alive(self, bot):
         activity = object()
         status = object()
-        bot._closing_event = mock.Mock()
-        shard = mock.Mock()
-        shard_obj = shard.return_value
-        shard_obj.is_alive = False
-
-        with mock.patch.object(aio, "first_completed", new=mock.AsyncMock()):
-            with mock.patch.object(shard_impl, "GatewayShardImpl", new=shard):
-                with pytest.raises(errors.GatewayError, match=r"shard 1 shut down immediately when starting"):
-                    await bot._start_one_shard(
-                        activity=activity,
-                        afk=True,
-                        idle_since=None,
-                        status=status,
-                        large_threshold=1000,
-                        shard_id=1,
-                        shard_count=3,
-                        url="https://some.website",
-                        closing_event=bot._closing_event,
-                    )
+        bot._shards = {}
+        shard_obj = mock.Mock(is_alive=False, start=mock.AsyncMock())
+
+        with mock.patch.object(shard_impl, "GatewayShardImpl", return_value=shard_obj):
+            with pytest.raises(RuntimeError, match=r"shard 1 shut down immediately when starting"):
+                await bot._start_one_shard(
+                    activity=activity,
+                    afk=True,
+                    idle_since=None,
+                    status=status,
+                    large_threshold=1000,
+                    shard_id=1,
+                    shard_count=3,
+                    url="https://some.website",
+                )
+
+        assert bot._shards == {}
+        shard_obj.close.assert_not_called()
+
+    @pytest.mark.parametrize("is_alive", [True, False])
+    @pytest.mark.asyncio()
+    async def test_start_one_shard_when_exception(self, bot, is_alive):
+        activity = object()
+        status = object()
+        bot._shards = {}
+        shard_obj = mock.Mock(
+            is_alive=is_alive,
+            start=mock.AsyncMock(side_effect=RuntimeError("exit in tests")),
+            close=mock.AsyncMock(),
+        )
+
+        with mock.patch.object(shard_impl, "GatewayShardImpl", return_value=shard_obj):
+            with pytest.raises(RuntimeError, match=r"exit in tests"):
+                await bot._start_one_shard(
+                    activity=activity,
+                    afk=True,
+                    idle_since=None,
+                    status=status,
+                    large_threshold=1000,
+                    shard_id=1,
+                    shard_count=3,
+                    url="https://some.website",
+                )
+
+        assert bot._shards == {}
+
+        if is_alive:
+            shard_obj.close.assert_awaited_once_with()
+        else:
+            shard_obj.close.assert_not_called()
diff --git a/tests/hikari/impl/test_buckets.py b/tests/hikari/impl/test_buckets.py
index 4984250e..23fab9be 100644
--- a/tests/hikari/impl/test_buckets.py
+++ b/tests/hikari/impl/test_buckets.py
@@ -161,6 +161,14 @@ async def test_start(self):
             mgr.start()
             assert mgr.gc_task is not None
 
+    @pytest.mark.asyncio()
+    async def test_start_when_already_started(self):
+        with buckets.RESTBucketManager(max_rate_limit=float("inf")) as mgr:
+            mock_task = mock.Mock()
+            mgr.gc_task = mock_task
+            mgr.start()
+            assert mgr.gc_task is mock_task
+
     @pytest.mark.asyncio()
     async def test_exit_closes(self):
         with mock.patch.object(buckets.RESTBucketManager, "close") as close:
diff --git a/tests/hikari/impl/test_entity_factory.py b/tests/hikari/impl/test_entity_factory.py
index c4d5d2d6..98e85d48 100644
--- a/tests/hikari/impl/test_entity_factory.py
+++ b/tests/hikari/impl/test_entity_factory.py
@@ -56,11 +56,6 @@
 from tests.hikari import hikari_test_helpers
 
 
-@pytest.fixture()
-def mock_app() -> traits.RESTAware:
-    return mock.MagicMock(traits.RESTAware)
-
-
 @pytest.fixture()
 def permission_overwrite_payload():
     return {"id": "4242", "type": 1, "allow": 65, "deny": 49152, "allow_new": "65", "deny_new": "49152"}
@@ -282,11 +277,17 @@ def test__deserialize_max_age_returns_null():
     assert entity_factory._deserialize_max_age(0) is None
 
 
-class TestGatewayGuildDefinition:
-    @pytest.fixture()
-    def entity_factory_impl(self, mock_app) -> entity_factory.EntityFactoryImpl:
-        return hikari_test_helpers.mock_class_namespace(entity_factory.EntityFactoryImpl, slots_=False)(mock_app)
+@pytest.fixture()
+def mock_app() -> traits.RESTAware:
+    return mock.Mock()
+
+
+@pytest.fixture()
+def entity_factory_impl(mock_app) -> entity_factory.EntityFactoryImpl:
+    return hikari_test_helpers.mock_class_namespace(entity_factory.EntityFactoryImpl, slots_=False)(mock_app)
 
+
+class TestGatewayGuildDefinition:
     def test_id_property(self, entity_factory_impl):
         guild_definition = entity_factory_impl.deserialize_gateway_guild({"id": "123123451234"})
 
@@ -640,12 +641,8 @@ def test_voice_states_returns_cached_values(self, entity_factory_impl):
 
 
 class TestEntityFactoryImpl:
-    @pytest.fixture()
-    def entity_factory_impl(self, mock_app) -> entity_factory.EntityFactoryImpl:
-        return entity_factory.EntityFactoryImpl(app=mock_app)
-
     def test_app(self, entity_factory_impl, mock_app):
-        assert entity_factory_impl._app is mock_app
+        assert entity_factory_impl.app is mock_app
 
     ######################
     # APPLICATION MODELS #
@@ -3653,11 +3650,12 @@ def test_deserialize_command_interaction_with_null_attributes(
         assert interaction.app_permissions is None
 
     @pytest.fixture()
-    def autocomplete_interaction_payload(self, user_payload, interaction_resolved_data_payload):
+    def autocomplete_interaction_payload(self, member_payload, user_payload, interaction_resolved_data_payload):
         return {
             "id": "3490190239012093",
             "type": 4,
             "guild_id": "43123123",
+            "member": member_payload,
             "data": {
                 "id": "43123123",
                 "name": "okokokok",
@@ -3667,8 +3665,8 @@ def autocomplete_interaction_payload(self, user_payload, interaction_resolved_da
                         "name": "options",
                         "type": 1,
                         "options": [
-                            {"name": "meat", "type": 4, "value": 123312},
-                            {"name": "yeet", "type": 3, "value": "ea", "focused": True},
+                            {"name": "meat", "type": 6, "value": 123312, "focused": True},
+                            {"name": "yeet", "type": 3, "value": "ea"},
                         ],
                     },
                 ],
@@ -3683,8 +3681,15 @@ def autocomplete_interaction_payload(self, user_payload, interaction_resolved_da
         }
 
     def test_deserialize_autocomplete_interaction(
-        self, entity_factory_impl, mock_app, autocomplete_interaction_payload
+        self,
+        entity_factory_impl,
+        mock_app,
+        member_payload,
+        autocomplete_interaction_payload,
+        interaction_resolved_data_payload,
     ):
+        entity_factory_impl._deserialize_interaction_member = mock.Mock()
+        entity_factory_impl._deserialize_resolved_option_data = mock.Mock()
         interaction = entity_factory_impl.deserialize_autocomplete_interaction(autocomplete_interaction_payload)
 
         assert interaction.app is mock_app
@@ -3695,9 +3700,9 @@ def test_deserialize_autocomplete_interaction(
         assert interaction.version == 69420
         assert interaction.channel_id == 49949494
         assert interaction.guild_id == 43123123
-        assert interaction.locale == "es-ES"
+        assert interaction.member is entity_factory_impl._deserialize_interaction_member.return_value
+        entity_factory_impl._deserialize_interaction_member.assert_called_once_with(member_payload, guild_id=43123123)
         assert interaction.locale is locales.Locale.ES_ES
-        assert interaction.guild_locale == "en-US"
         assert interaction.guild_locale is locales.Locale.EN_US
 
         # AutocompleteInteractionOption
@@ -3711,9 +3716,10 @@ def test_deserialize_autocomplete_interaction(
         sub_option1 = option.options[0]
         assert sub_option1.name == "meat"
         assert sub_option1.value == 123312
-        assert sub_option1.type is commands.OptionType.INTEGER
+        assert isinstance(sub_option1.value, snowflakes.Snowflake)
+        assert sub_option1.type is commands.OptionType.USER
         assert sub_option1.options is None
-        assert sub_option1.is_focused is False
+        assert sub_option1.is_focused is True
         assert isinstance(sub_option1, command_interactions.CommandInteractionOption)
 
         sub_option2 = option.options[1]
@@ -3721,7 +3727,7 @@ def test_deserialize_autocomplete_interaction(
         assert sub_option2.value == "ea"
         assert sub_option2.type is commands.OptionType.STRING
         assert sub_option2.options is None
-        assert sub_option2.is_focused is True
+        assert sub_option2.is_focused is False
         assert isinstance(sub_option2, command_interactions.AutocompleteInteractionOption)
         assert isinstance(option, command_interactions.AutocompleteInteractionOption)
 
@@ -3732,14 +3738,19 @@ def test_deserialize_autocomplete_interaction_with_null_fields(
     ):
         del autocomplete_interaction_payload["guild_locale"]
         del autocomplete_interaction_payload["guild_id"]
+        del autocomplete_interaction_payload["member"]
+
+        entity_factory_impl.deserialize_user = mock.Mock()
 
         interaction = entity_factory_impl.deserialize_autocomplete_interaction(autocomplete_interaction_payload)
 
         assert interaction.guild_id is None
         assert interaction.member is None
-        assert interaction.user == entity_factory_impl.deserialize_user(user_payload)
         assert interaction.guild_locale is None
 
+        assert interaction.user is entity_factory_impl.deserialize_user.return_value
+        entity_factory_impl.deserialize_user.assert_called_once_with(user_payload)
+
     @pytest.mark.parametrize(
         ("type_", "fn"),
         [
@@ -3848,6 +3859,23 @@ def test_deserialize_context_menu_command(self, entity_factory_impl, context_men
         assert command.is_dm_enabled is True
         assert command.version == 123321123
 
+    def test_deserialize_context_menu_command_with_guild_id(
+        self,
+        entity_factory_impl,
+        context_menu_command_payload,
+    ):
+        command = entity_factory_impl.deserialize_command(context_menu_command_payload, guild_id=123)
+        assert isinstance(command, commands.ContextMenuCommand)
+
+        assert command.id == 1231231231
+        assert command.application_id == 12354123
+        assert command.guild_id == 123
+        assert command.type == commands.CommandType.USER
+        assert command.name == "good name"
+        assert command.default_member_permissions == permission_models.Permissions.ADMINISTRATOR
+        assert command.is_dm_enabled is True
+        assert command.version == 123321123
+
     def test_deserialize_context_menu_command_with_with_null_and_unset_values(
         self, entity_factory_impl, context_menu_command_payload
     ):
@@ -5303,6 +5331,11 @@ def test_deserialize_scheduled_external_event_with_null_fields(
         scheduled_external_event_payload["description"] = None
         scheduled_external_event_payload["image"] = None
 
+        event = entity_factory_impl.deserialize_scheduled_external_event(scheduled_external_event_payload)
+
+        assert event.description is None
+        assert event.image_hash is None
+
     def test_deserialize_scheduled_external_event_with_undefined_fields(
         self,
         entity_factory_impl: entity_factory.EntityFactoryImpl,
@@ -5314,6 +5347,13 @@ def test_deserialize_scheduled_external_event_with_undefined_fields(
         del scheduled_external_event_payload["image"]
         del scheduled_external_event_payload["user_count"]
 
+        event = entity_factory_impl.deserialize_scheduled_external_event(scheduled_external_event_payload)
+
+        assert event.description is None
+        assert event.image_hash is None
+        assert event.creator is None
+        assert event.user_count is None
+
     @pytest.fixture()
     def scheduled_stage_event_payload(self, user_payload: typing.Dict[str, typing.Any]) -> typing.Dict[str, typing.Any]:
         return {
@@ -5492,6 +5532,10 @@ def test_deserialize_scheduled_event_returns_right_type(
 
             assert isinstance(result, cls)
 
+    def test_deserialize_scheduled_event_when_unknown(self, entity_factory_impl: entity_factory.EntityFactoryImpl):
+        with pytest.raises(errors.UnrecognisedEntityError):
+            entity_factory_impl.deserialize_scheduled_event({"entity_type": -1})
+
     @pytest.fixture()
     def scheduled_event_user_payload(
         self,
diff --git a/tests/hikari/impl/test_event_manager_base.py b/tests/hikari/impl/test_event_manager_base.py
index 680a7063..39de4f63 100644
--- a/tests/hikari/impl/test_event_manager_base.py
+++ b/tests/hikari/impl/test_event_manager_base.py
@@ -750,7 +750,7 @@ def test_get_listeners_monomorphic_and_no_results(self, event_manager):
             member_events.MemberDeleteEvent: ["coroutine4", "coroutine5"],
         }
 
-        assert event_manager.get_listeners(member_events.MemberEvent, polymorphic=False) == []
+        assert event_manager.get_listeners(member_events.MemberEvent, polymorphic=False) == ()
 
     def test_get_listeners_monomorphic_and_results(self, event_manager):
         event_manager._listeners = {
diff --git a/tests/hikari/impl/test_interaction_server.py b/tests/hikari/impl/test_interaction_server.py
index 8650b707..172cf735 100644
--- a/tests/hikari/impl/test_interaction_server.py
+++ b/tests/hikari/impl/test_interaction_server.py
@@ -520,7 +520,7 @@ async def test_close(self, mock_interaction_server: interaction_server_impl.Inte
         mock_runner.shutdown.assert_awaited_once()
         mock_runner.cleanup.assert_awaited_once()
         mock_event.set.assert_called_once()
-        assert mock_interaction_server._is_closing is True
+        assert mock_interaction_server._is_closing is False
 
     @pytest.mark.asyncio()
     async def test_close_when_closing(self, mock_interaction_server: interaction_server_impl.InteractionServer):
@@ -825,7 +825,6 @@ async def test_start(self, mock_interaction_server: interaction_server_impl.Inte
         with stack:
             await mock_interaction_server.start(
                 backlog=123123,
-                enable_signal_handlers=False,
                 host="hoototototo",
                 port=123123123,
                 path="hshshshshsh",
@@ -841,13 +840,13 @@ async def test_start(self, mock_interaction_server: interaction_server_impl.Inte
                 [aiohttp.web.post("/", mock_interaction_server.aiohttp_hook)]
             )
             aiohttp.web_runner.AppRunner.assert_called_once_with(
-                aiohttp.web.Application.return_value, handle_signals=False, access_log=interaction_server_impl._LOGGER
+                aiohttp.web.Application.return_value, access_log=interaction_server_impl._LOGGER
             )
             aiohttp.web_runner.AppRunner.return_value.setup.assert_awaited_once()
             aiohttp.web.TCPSite.assert_called_once_with(
                 aiohttp.web_runner.AppRunner.return_value,
                 "hoototototo",
-                123123123,
+                port=123123123,
                 shutdown_timeout=3232.3232,
                 ssl_context=mock_context,
                 backlog=123123,
@@ -892,7 +891,7 @@ async def test_start_with_default_behaviour(
                 [aiohttp.web.post("/", mock_interaction_server.aiohttp_hook)]
             )
             aiohttp.web_runner.AppRunner.assert_called_once_with(
-                aiohttp.web.Application.return_value, handle_signals=True, access_log=interaction_server_impl._LOGGER
+                aiohttp.web.Application.return_value, access_log=interaction_server_impl._LOGGER
             )
             aiohttp.web_runner.AppRunner.return_value.setup.assert_awaited_once()
             aiohttp.web.TCPSite.assert_called_once_with(
@@ -925,7 +924,7 @@ async def test_start_with_default_behaviour_and_not_main_thread(
                 [aiohttp.web.post("/", mock_interaction_server.aiohttp_hook)]
             )
             aiohttp.web_runner.AppRunner.assert_called_once_with(
-                aiohttp.web.Application.return_value, handle_signals=False, access_log=interaction_server_impl._LOGGER
+                aiohttp.web.Application.return_value, access_log=interaction_server_impl._LOGGER
             )
             aiohttp.web_runner.AppRunner.return_value.setup.assert_awaited_once()
             aiohttp.web.TCPSite.assert_called_once_with(
@@ -955,7 +954,7 @@ async def test_start_with_multiple_hosts(self, mock_interaction_server: interact
                 [aiohttp.web.post("/", mock_interaction_server.aiohttp_hook)]
             )
             aiohttp.web_runner.AppRunner.assert_called_once_with(
-                aiohttp.web.Application.return_value, handle_signals=True, access_log=interaction_server_impl._LOGGER
+                aiohttp.web.Application.return_value, access_log=interaction_server_impl._LOGGER
             )
             aiohttp.web_runner.AppRunner.return_value.setup.assert_awaited_once()
             aiohttp.web.TCPSite.assert_has_calls(
@@ -963,7 +962,7 @@ async def test_start_with_multiple_hosts(self, mock_interaction_server: interact
                     mock.call(
                         aiohttp.web_runner.AppRunner.return_value,
                         "123",
-                        453123,
+                        port=453123,
                         shutdown_timeout=60.0,
                         ssl_context=mock_context,
                         backlog=128,
@@ -973,7 +972,7 @@ async def test_start_with_multiple_hosts(self, mock_interaction_server: interact
                     mock.call(
                         aiohttp.web_runner.AppRunner.return_value,
                         "4312",
-                        453123,
+                        port=453123,
                         shutdown_timeout=60.0,
                         ssl_context=mock_context,
                         backlog=128,
@@ -984,6 +983,39 @@ async def test_start_with_multiple_hosts(self, mock_interaction_server: interact
             )
             aiohttp.web.TCPSite.return_value.start.assert_has_awaits([mock.call(), mock.call()])
 
+    @pytest.mark.asyncio()
+    async def test_start_when_no_tcp_sites(self, mock_interaction_server: interaction_server_impl.InteractionServer):
+        mock_socket = object()
+        mock_context = object()
+        stack = contextlib.ExitStack()
+        stack.enter_context(mock.patch.object(aiohttp.web, "TCPSite", return_value=mock.AsyncMock()))
+        stack.enter_context(mock.patch.object(aiohttp.web_runner, "AppRunner", return_value=mock.AsyncMock()))
+        stack.enter_context(mock.patch.object(aiohttp.web, "Application"))
+        stack.enter_context(mock.patch.object(aiohttp.web, "SockSite", return_value=mock.AsyncMock()))
+        stack.enter_context(mock.patch.object(aiohttp.web, "UnixSite", return_value=mock.AsyncMock()))
+
+        with stack:
+            await mock_interaction_server.start(ssl_context=mock_context, socket=mock_socket)
+
+            aiohttp.web.Application.assert_called_once_with()
+            aiohttp.web.Application.return_value.add_routes.assert_called_once_with(
+                [aiohttp.web.post("/", mock_interaction_server.aiohttp_hook)]
+            )
+            aiohttp.web_runner.AppRunner.assert_called_once_with(
+                aiohttp.web.Application.return_value, access_log=interaction_server_impl._LOGGER
+            )
+            aiohttp.web_runner.AppRunner.return_value.setup.assert_awaited_once()
+            aiohttp.web.TCPSite.assert_not_called()
+            aiohttp.web.UnixSite.assert_not_called()
+            aiohttp.web.SockSite.assert_called_once_with(
+                aiohttp.web_runner.AppRunner.return_value,
+                mock_socket,
+                shutdown_timeout=60.0,
+                ssl_context=mock_context,
+                backlog=128,
+            )
+            aiohttp.web.SockSite.return_value.start.assert_awaited_once()
+
     @pytest.mark.asyncio()
     async def test_start_when_already_running(self, mock_interaction_server: interaction_server_impl.InteractionServer):
         mock_interaction_server._server = object()
diff --git a/tests/hikari/impl/test_rest_bot.py b/tests/hikari/impl/test_rest_bot.py
index ab2120c0..3ac09533 100644
--- a/tests/hikari/impl/test_rest_bot.py
+++ b/tests/hikari/impl/test_rest_bot.py
@@ -34,6 +34,7 @@
 from hikari.impl import rest as rest_impl
 from hikari.impl import rest_bot as rest_bot_impl
 from hikari.internal import aio
+from hikari.internal import signals
 from hikari.internal import ux
 from tests.hikari import hikari_test_helpers
 
@@ -162,7 +163,7 @@ def test___init___parses_string_public_key(self):
         stack.enter_context(mock.patch.object(interaction_server_impl, "InteractionServer"))
 
         with stack:
-            result = cls("token", "token_type", "6f66646f646f646f6f")
+            result = cls(object(), "token_type", "6f66646f646f646f6f")
 
             interaction_server_impl.InteractionServer.assert_called_once_with(
                 entity_factory=result.entity_factory, public_key=b"ofdododoo", rest_client=result.rest
@@ -247,7 +248,7 @@ async def test_close(self, mock_rest_bot, mock_interaction_server, mock_rest_cli
         mock_interaction_server.close.assert_awaited_once()
         mock_rest_client.close.assert_awaited_once()
         close_event.set.assert_called_once()
-        assert mock_rest_bot._is_closing is True
+        assert mock_rest_bot._is_closing is False
 
     @pytest.mark.asyncio()
     async def test_close_when_is_closing(self, mock_rest_bot, mock_interaction_server, mock_rest_client):
@@ -262,6 +263,7 @@ async def test_close_when_is_closing(self, mock_rest_bot, mock_interaction_serve
         mock_rest_client.close.assert_not_called()
         mock_rest_bot._close_event.set.assert_not_called()
         mock_rest_bot.join.assert_awaited_once()
+        assert mock_rest_bot._is_closing is True
 
     @pytest.mark.asyncio()
     async def test_close_when_inactive(self, mock_rest_bot):
@@ -301,30 +303,39 @@ def test_run(self, mock_rest_bot):
         mock_rest_bot.join = mock.AsyncMock()
 
         with mock.patch.object(ux, "check_for_updates") as check_for_updates:
-            mock_rest_bot.run(
-                asyncio_debug=False,
-                backlog=321,
-                check_for_updates=False,
-                close_loop=False,
-                close_passed_executor=False,
-                coroutine_tracking_depth=32123,
-                enable_signal_handlers=True,
-                host="192.168.1.102",
-                path="pathathath",
-                port=4554,
-                reuse_address=True,
-                reuse_port=False,
-                shutdown_timeout=534.534,
-                socket=mock_socket,
-                ssl_context=mock_context,
-            )
-
-            check_for_updates.assert_not_called()
+            with mock.patch.object(
+                signals, "handle_interrupts", return_value=hikari_test_helpers.ContextManagerMock()
+            ) as handle_interrupts:
+                mock_rest_bot.run(
+                    asyncio_debug=False,
+                    backlog=321,
+                    check_for_updates=False,
+                    close_loop=False,
+                    close_passed_executor=False,
+                    coroutine_tracking_depth=32123,
+                    enable_signal_handlers=True,
+                    propagate_interrupts=True,
+                    host="192.168.1.102",
+                    path="pathathath",
+                    port=4554,
+                    reuse_address=True,
+                    reuse_port=False,
+                    shutdown_timeout=534.534,
+                    socket=mock_socket,
+                    ssl_context=mock_context,
+                )
+
+        check_for_updates.assert_not_called()
+        handle_interrupts.assert_called_once_with(
+            enabled=True,
+            loop=asyncio.get_event_loop_policy().get_event_loop(),
+            propagate_interrupts=True,
+        )
+        handle_interrupts.return_value.assert_used_once()
 
         mock_rest_bot.start.assert_awaited_once_with(
             backlog=321,
             check_for_updates=False,
-            enable_signal_handlers=True,
             host="192.168.1.102",
             path="pathathath",
             port=4554,
@@ -337,12 +348,23 @@ def test_run(self, mock_rest_bot):
         mock_rest_bot.join.assert_awaited_once()
         assert asyncio.get_event_loop_policy().get_event_loop().is_closed() is False
 
+    def test_run_when_close_loop(self, mock_rest_bot):
+        mock_rest_bot.start = mock.Mock()
+        mock_rest_bot.join = mock.Mock()
+
+        with mock.patch.object(aio, "get_or_make_loop") as get_or_make_loop:
+            with mock.patch.object(aio, "destroy_loop") as destroy_loop:
+                with mock.patch.object(rest_bot_impl, "_LOGGER") as logger:
+                    mock_rest_bot.run(close_loop=True)
+
+        destroy_loop.assert_called_once_with(get_or_make_loop.return_value, logger)
+
     def test_run_when_asyncio_debug(self, mock_rest_bot):
         mock_rest_bot.start = mock.Mock()
         mock_rest_bot.join = mock.Mock()
 
         with mock.patch.object(aio, "get_or_make_loop") as get_or_make_loop:
-            mock_rest_bot.run(asyncio_debug=True)
+            mock_rest_bot.run(asyncio_debug=True, close_loop=False)
 
         get_or_make_loop.return_value.set_debug.assert_called_once_with(True)
 
@@ -354,7 +376,7 @@ def test_run_with_coroutine_tracking_depth(self, mock_rest_bot):
             with mock.patch.object(
                 sys, "set_coroutine_origin_tracking_depth", side_effect=AttributeError, create=True
             ) as set_tracking_depth:
-                mock_rest_bot.run(coroutine_tracking_depth=42)
+                mock_rest_bot.run(coroutine_tracking_depth=42, close_loop=False)
 
         set_tracking_depth.assert_called_once_with(42)
 
@@ -427,7 +449,6 @@ async def test_start(self, mock_rest_bot, mock_interaction_server, mock_rest_cli
             await mock_rest_bot.start(
                 backlog=34123,
                 check_for_updates=False,
-                enable_signal_handlers=False,
                 host="hostostosot",
                 port=123123123,
                 path="patpatpapt",
@@ -442,7 +463,6 @@ async def test_start(self, mock_rest_bot, mock_interaction_server, mock_rest_cli
 
         mock_interaction_server.start.assert_awaited_once_with(
             backlog=34123,
-            enable_signal_handlers=False,
             host="hostostosot",
             port=123123123,
             path="patpatpapt",
@@ -465,7 +485,6 @@ async def test_start_checks_for_update(self, mock_rest_bot, mock_http_settings,
             await mock_rest_bot.start(
                 backlog=34123,
                 check_for_updates=True,
-                enable_signal_handlers=False,
                 host="hostostosot",
                 port=123123123,
                 path="patpatpapt",
diff --git a/tests/hikari/impl/test_shard.py b/tests/hikari/impl/test_shard.py
index 9e50f045..863561a0 100644
--- a/tests/hikari/impl/test_shard.py
+++ b/tests/hikari/impl/test_shard.py
@@ -33,11 +33,13 @@
 from hikari import errors
 from hikari import intents
 from hikari import presences
-from hikari import undefined
 from hikari.impl import config
 from hikari.impl import shard
+from hikari.internal import aio
+from hikari.internal import data_binding
+from hikari.internal import net
 from hikari.internal import time
-from tests.hikari import client_session_stub
+from hikari.internal import ux
 from tests.hikari import hikari_test_helpers
 
 
@@ -50,6 +52,25 @@ def test_log_filterer():
     )
 
 
+def test__serialize_activity_when_activity_is_None():
+    assert shard._serialize_activity(None) is None
+
+
+def test__serialize_activity_when_activity_is_not_None():
+    activity = mock.Mock(type="0", url="https://some.url")
+    activity.name = "some name"  # This has to be set separate because if not, its set as the mock's name
+    assert shard._serialize_activity(activity) == {"name": "some name", "type": 0, "url": "https://some.url"}
+
+
+def test__serialize_datetime_when_datetime_is_None():
+    assert shard._serialize_datetime(None) is None
+
+
+def test__serialize_datetime_when_datetime_is_not_None():
+    dt = datetime.datetime(2004, 11, 22, tzinfo=datetime.timezone.utc)
+    assert shard._serialize_datetime(dt) == 1101081600000
+
+
 @pytest.fixture()
 def http_settings():
     return mock.Mock(spec_set=config.HTTPSettings)
@@ -76,41 +97,64 @@ def __init__(
 class TestGatewayTransport:
     @pytest.fixture()
     def transport_impl(self):
-        with mock.patch.object(aiohttp.ClientWebSocketResponse, "__init__"):
-            transport = shard._GatewayTransport()
-            transport.logger = mock.Mock(isEnabledFor=mock.Mock(return_value=True))
-            transport.log_filterer = mock.Mock()
-            yield transport
+        return shard._GatewayTransport(
+            ws=mock.Mock(),
+            exit_stack=mock.AsyncMock(),
+            logger=mock.Mock(),
+            log_filterer=mock.Mock(),
+            transport_compression=True,
+        )
 
-    def test__init__calls_super(self):
-        with mock.patch.object(aiohttp.ClientWebSocketResponse, "__init__") as init:
-            shard._GatewayTransport("arg1", "arg2", some_kwarg="kwarg1")
+    def test_init_when_transport_compression(self):
+        transport = shard._GatewayTransport(
+            ws=mock.Mock(),
+            exit_stack=mock.AsyncMock(),
+            logger=mock.Mock(),
+            log_filterer=mock.Mock(),
+            transport_compression=True,
+        )
+
+        assert transport._receive_and_check == transport._receive_and_check_zlib
 
-        init.assert_called_once_with("arg1", "arg2", some_kwarg="kwarg1")
+    def test_init_when_no_transport_compression(self):
+        transport = shard._GatewayTransport(
+            ws=mock.Mock(),
+            exit_stack=mock.AsyncMock(),
+            logger=mock.Mock(isEnabledFor=mock.Mock(return_value=False)),
+            log_filterer=mock.Mock(),
+            transport_compression=False,
+        )
+
+        assert transport._receive_and_check == transport._receive_and_check_text
 
     @pytest.mark.asyncio()
-    async def test_send_close_when_not_closed_nor_closing_logs(self, transport_impl):
-        transport_impl.sent_close = False
+    async def test_send_close(self, transport_impl):
+        transport_impl._sent_close = False
 
-        with mock.patch.object(aiohttp.ClientWebSocketResponse, "close", new=mock.Mock()) as close:
-            with mock.patch.object(asyncio, "wait_for", return_value=mock.AsyncMock()) as wait_for:
+        with mock.patch.object(asyncio, "wait_for", return_value=mock.AsyncMock()) as wait_for:
+            with mock.patch.object(asyncio, "sleep") as sleep:
                 await transport_impl.send_close(code=1234, message=b"some message")
 
-        wait_for.assert_awaited_once_with(close.return_value, timeout=5)
-        close.assert_called_once_with(code=1234, message=b"some message")
+        wait_for.assert_awaited_once_with(transport_impl._ws.close.return_value, timeout=5)
+        transport_impl._ws.close.assert_called_once_with(code=1234, message=b"some message")
+        transport_impl._exit_stack.aclose.assert_awaited_once_with()
+        sleep.assert_awaited_once_with(0.25)
 
     @pytest.mark.asyncio()
     async def test_send_close_when_TimeoutError(self, transport_impl):
-        transport_impl.sent_close = False
+        transport_impl._sent_close = False
+        transport_impl._ws.close.side_effect = asyncio.TimeoutError
 
-        with mock.patch.object(aiohttp.ClientWebSocketResponse, "close", side_effect=asyncio.TimeoutError) as close:
+        with mock.patch.object(asyncio, "sleep") as sleep:
             await transport_impl.send_close(code=1234, message=b"some message")
 
-        close.assert_called_once_with(code=1234, message=b"some message")
+        transport_impl._ws.close.assert_called_once_with(code=1234, message=b"some message")
+        transport_impl._exit_stack.aclose.assert_awaited_once_with()
+        sleep.assert_awaited_once_with(0.25)
 
     @pytest.mark.asyncio()
     async def test_send_close_when_already_sent(self, transport_impl):
-        transport_impl.sent_close = True
+        transport_impl._sent_close = True
 
         with mock.patch.object(aiohttp.ClientWebSocketResponse, "close", side_effect=asyncio.TimeoutError) as close:
             await transport_impl.send_close(code=1234, message=b"some message")
@@ -120,28 +164,41 @@ async def test_send_close_when_already_sent(self, transport_impl):
     @pytest.mark.asyncio()
     @pytest.mark.parametrize("trace", [True, False])
     async def test_receive_json(self, transport_impl, trace):
-        transport_impl._receive_and_check = mock.AsyncMock(return_value="{'json_response': null}")
-        transport_impl.logger.isEnabledFor.return_value = trace
-        mock_loads = mock.Mock(return_value={"json_response": None})
+        transport_impl._receive_and_check = mock.AsyncMock()
+        transport_impl._log_payload = trace
 
-        assert await transport_impl.receive_json(loads=mock_loads, timeout=69) == {"json_response": None}
+        with mock.patch.object(data_binding, "load_json") as load_json:
+            assert await transport_impl.receive_json() == load_json.return_value
 
-        transport_impl._receive_and_check.assert_awaited_once_with(69)
-        mock_loads.assert_called_once_with("{'json_response': null}")
+        transport_impl._receive_and_check.assert_awaited_once_with()
+        load_json.assert_called_once_with(transport_impl._receive_and_check.return_value)
 
     @pytest.mark.asyncio()
     @pytest.mark.parametrize("trace", [True, False])
     async def test_send_json(self, transport_impl, trace):
-        transport_impl.send_str = mock.AsyncMock()
-        transport_impl.logger.isEnabledFor.return_value = trace
-        mock_dumps = mock.Mock(return_value="{'json_send': null}")
+        transport_impl._ws.send_str = mock.AsyncMock()
+        transport_impl._log_payload = trace
 
-        await transport_impl.send_json({"json_send": None}, 420, dumps=mock_dumps)
+        with mock.patch.object(data_binding, "dump_json") as dump_json:
+            await transport_impl.send_json({"json_send": None})
 
-        transport_impl.send_str.assert_awaited_once_with("{'json_send': null}", 420)
-        mock_dumps.assert_called_once_with({"json_send": None})
+        transport_impl._ws.send_str.assert_awaited_once_with(dump_json.return_value)
+        dump_json.assert_called_once_with({"json_send": None})
 
     @pytest.mark.asyncio()
+    async def test__handle_other_message_when_TEXT(self, transport_impl):
+        stub_response = StubResponse(type=aiohttp.WSMsgType.TEXT)
+
+        with pytest.raises(errors.GatewayError, match="Unexpected message type received TEXT, expected BINARY"):
+            transport_impl._handle_other_message(stub_response)
+
+    @pytest.mark.asyncio()
+    async def test__handle_other_message_when_BINARY(self, transport_impl):
+        stub_response = StubResponse(type=aiohttp.WSMsgType.BINARY)
+
+        with pytest.raises(errors.GatewayError, match="Unexpected message type received BINARY, expected TEXT"):
+            transport_impl._handle_other_message(stub_response)
+
     @pytest.mark.parametrize(
         "code",
         [
@@ -153,373 +210,235 @@ async def test_send_json(self, transport_impl, trace):
             errors.ShardCloseCode.RATE_LIMITED,
         ],
     )
-    async def test__handle_other_message_when_message_type_is_CLOSE_and_should_reconnect(self, code, transport_impl):
+    def test__handle_other_message_when_message_type_is_CLOSE_and_should_reconnect(self, code, transport_impl):
         stub_response = StubResponse(type=aiohttp.WSMsgType.CLOSE, extra="some error extra", data=code)
 
         with pytest.raises(errors.GatewayServerClosedConnectionError) as exinfo:
-            await transport_impl._handle_other_message(stub_response)
+            transport_impl._handle_other_message(stub_response)
 
         exception = exinfo.value
         assert exception.reason == "some error extra"
         assert exception.code == int(code)
         assert exception.can_reconnect is True
 
-    @pytest.mark.asyncio()
     @pytest.mark.parametrize(
         "code",
         [*range(4010, 4020), 5000],
     )
-    async def test__handle_other_message_when_message_type_is_CLOSE_and_should_not_reconnect(
-        self, code, transport_impl
-    ):
+    def test__handle_other_message_when_message_type_is_CLOSE_and_should_not_reconnect(self, code, transport_impl):
         stub_response = StubResponse(type=aiohttp.WSMsgType.CLOSE, extra="don't reconnect", data=code)
 
         with pytest.raises(errors.GatewayServerClosedConnectionError) as exinfo:
-            await transport_impl._handle_other_message(stub_response)
+            transport_impl._handle_other_message(stub_response)
 
         exception = exinfo.value
         assert exception.reason == "don't reconnect"
         assert exception.code == int(code)
         assert exception.can_reconnect is False
 
-    @pytest.mark.asyncio()
-    async def test__handle_other_message_when_message_type_is_CLOSING(self, transport_impl):
+    def test__handle_other_message_when_message_type_is_CLOSING(self, transport_impl):
         stub_response = StubResponse(type=aiohttp.WSMsgType.CLOSING)
 
         with pytest.raises(errors.GatewayError, match="Socket has closed"):
-            await transport_impl._handle_other_message(stub_response)
+            transport_impl._handle_other_message(stub_response)
 
-    @pytest.mark.asyncio()
-    async def test__handle_other_message_when_message_type_is_CLOSED(self, transport_impl):
+    def test__handle_other_message_when_message_type_is_CLOSED(self, transport_impl):
         stub_response = StubResponse(type=aiohttp.WSMsgType.CLOSED)
 
         with pytest.raises(errors.GatewayError, match="Socket has closed"):
-            await transport_impl._handle_other_message(stub_response)
+            transport_impl._handle_other_message(stub_response)
 
-    @pytest.mark.asyncio()
-    async def test__handle_other_message_when_message_type_is_unknown(self, transport_impl):
+    def test__handle_other_message_when_message_type_is_unknown(self, transport_impl):
         stub_response = mock.AsyncMock(return_value=StubResponse(type=aiohttp.WSMsgType.ERROR))
-        transport_impl.exception = mock.Mock(return_value=Exception("some error"))
+        exception = Exception("some error")
+        transport_impl._ws.exception = mock.Mock(return_value=exception)
 
         with pytest.raises(errors.GatewayError, match="Unexpected websocket exception from gateway") as exc_info:
-            await transport_impl._handle_other_message(stub_response)
+            transport_impl._handle_other_message(stub_response)
 
-        assert exc_info.value.__cause__ is transport_impl.exception.return_value
+        assert exc_info.value.__cause__ is exception
 
     @pytest.mark.asyncio()
-    async def test__receive_and_check_when_message_type_is_BINARY(self, transport_impl):
-        response = StubResponse(type=aiohttp.WSMsgType.BINARY, data=b"some initial data")
-        transport_impl.receive = mock.AsyncMock(return_value=response)
-        transport_impl._receive_and_check_complete_zlib_package = mock.AsyncMock()
+    async def test__receive_and_check_text_when_message_type_is_TEXT(self, transport_impl):
+        transport_impl._ws.receive = mock.AsyncMock(
+            return_value=StubResponse(type=aiohttp.WSMsgType.TEXT, data="some text")
+        )
 
-        result = await transport_impl._receive_and_check(10)
+        assert await transport_impl._receive_and_check_text() == "some text"
 
-        assert result is transport_impl._receive_and_check_complete_zlib_package.return_value
-        transport_impl.receive.assert_awaited_once_with(10)
-        transport_impl._receive_and_check_complete_zlib_package.assert_awaited_once_with(b"some initial data", 10)
+        transport_impl._ws.receive.assert_awaited_once_with()
 
     @pytest.mark.asyncio()
-    async def test__receive_and_check_when_message_type_is_BINARY_and_the_full_payload(self, transport_impl):
-        response = StubResponse(type=aiohttp.WSMsgType.BINARY, data=b"some initial data\x00\x00\xff\xff")
-        transport_impl.receive = mock.AsyncMock(return_value=response)
-        transport_impl.zlib = mock.Mock(decompress=mock.Mock(return_value=b"aaaaaaaaaaaaaaaaaa"))
+    async def test__receive_and_check_text_when_message_type_is_unknown(self, transport_impl):
+        mock_exception = errors.GatewayError("aye")
+        transport_impl._ws.receive = mock.AsyncMock(return_value=StubResponse(type=aiohttp.WSMsgType.BINARY))
 
-        assert await transport_impl._receive_and_check(10) == "aaaaaaaaaaaaaaaaaa"
+        with mock.patch.object(
+            shard._GatewayTransport, "_handle_other_message", side_effect=mock_exception
+        ) as handle_other_message:
+            with pytest.raises(errors.GatewayError) as exc_info:
+                await transport_impl._receive_and_check_text()
 
-        transport_impl.receive.assert_awaited_once_with(10)
-        transport_impl.zlib.decompress.assert_called_once_with(response.data)
+        assert exc_info.value is mock_exception
+        transport_impl._ws.receive.assert_awaited_once_with()
+        handle_other_message.assert_called_once_with(transport_impl._ws.receive.return_value)
 
     @pytest.mark.asyncio()
-    async def test__receive_and_check_when_message_type_is_TEXT(self, transport_impl):
-        transport_impl.receive = mock.AsyncMock(
-            return_value=StubResponse(type=aiohttp.WSMsgType.TEXT, data="some text")
-        )
+    async def test__receive_and_check_zlib_when_message_type_is_BINARY(self, transport_impl):
+        response = StubResponse(type=aiohttp.WSMsgType.BINARY, data=b"some initial data")
+        transport_impl._ws.receive = mock.AsyncMock(return_value=response)
 
-        assert await transport_impl._receive_and_check(10) == "some text"
+        with mock.patch.object(
+            shard._GatewayTransport, "_receive_and_check_complete_zlib_package"
+        ) as receive_and_check_complete_zlib_package:
+            assert (
+                await transport_impl._receive_and_check_zlib() is receive_and_check_complete_zlib_package.return_value
+            )
 
-        transport_impl.receive.assert_awaited_once_with(10)
+        transport_impl._ws.receive.assert_awaited_once_with()
+        receive_and_check_complete_zlib_package.assert_awaited_once_with(b"some initial data")
 
     @pytest.mark.asyncio()
-    async def test__receive_and_check_when_message_type_is_unknown(self, transport_impl):
-        mock_exception = errors.GatewayError("aye")
-        transport_impl.receive = mock.AsyncMock(return_value=StubResponse(type=aiohttp.WSMsgType.ERROR))
-        transport_impl._handle_other_message = mock.Mock(side_effect=mock_exception)
+    async def test__receive_and_check_zlib_when_message_type_is_BINARY_and_the_full_payload(self, transport_impl):
+        response = StubResponse(type=aiohttp.WSMsgType.BINARY, data=b"some initial data\x00\x00\xff\xff")
+        transport_impl._ws.receive = mock.AsyncMock(return_value=response)
+        transport_impl._zlib = mock.Mock(decompress=mock.Mock(return_value=b"aaaaaaaaaaaaaaaaaa"))
 
-        with pytest.raises(errors.GatewayError) as exc_info:
-            await transport_impl._receive_and_check(10)
+        assert await transport_impl._receive_and_check_zlib() == "aaaaaaaaaaaaaaaaaa"
 
-        assert exc_info.value is mock_exception
-        transport_impl.receive.assert_awaited_once_with(10)
-        transport_impl._handle_other_message.assert_called_once_with(transport_impl.receive.return_value)
+        transport_impl._ws.receive.assert_awaited_once_with()
+        transport_impl._zlib.decompress.assert_called_once_with(response.data)
 
     @pytest.mark.asyncio()
-    async def test__receive_and_check_complete_zlib_package_when_TEXT(self, transport_impl):
-        response = StubResponse(type=aiohttp.WSMsgType.TEXT, data="not binary")
-        transport_impl.receive = mock.AsyncMock(return_value=response)
+    async def test__receive_and_check_zlib_when_message_type_is_unknown(self, transport_impl):
+        mock_exception = errors.GatewayError("aye")
+        transport_impl._ws.receive = mock.AsyncMock(return_value=StubResponse(type=aiohttp.WSMsgType.TEXT))
 
-        with pytest.raises(errors.GatewayError, match="Unexpected message type received TEXT, expected BINARY"):
-            await transport_impl._receive_and_check_complete_zlib_package(b"some", 10)
+        with mock.patch.object(
+            shard._GatewayTransport, "_handle_other_message", side_effect=mock_exception
+        ) as handle_other_message:
+            with pytest.raises(errors.GatewayError) as exc_info:
+                await transport_impl._receive_and_check_zlib()
 
-        transport_impl.receive.assert_awaited_with(10)
+        assert exc_info.value is mock_exception
+        transport_impl._ws.receive.assert_awaited_once_with()
+        handle_other_message.assert_called_once_with(transport_impl._ws.receive.return_value)
 
     @pytest.mark.asyncio()
     async def test__receive_and_check_complete_zlib_package_for_unexpected_message_type(self, transport_impl):
         mock_exception = errors.GatewayError("aye")
-        response = StubResponse(type=aiohttp.WSMsgType.CLOSE)
-        transport_impl.receive = mock.AsyncMock(return_value=response)
-        transport_impl._handle_other_message = mock.Mock(side_effect=mock_exception)
+        response = StubResponse(type=aiohttp.WSMsgType.TEXT)
+        transport_impl._ws.receive = mock.AsyncMock(return_value=response)
 
-        with pytest.raises(errors.GatewayError) as exc_info:
-            await transport_impl._receive_and_check_complete_zlib_package(b"some", 10)
+        with mock.patch.object(
+            shard._GatewayTransport, "_handle_other_message", side_effect=mock_exception
+        ) as handle_other_message:
+            with pytest.raises(errors.GatewayError) as exc_info:
+                await transport_impl._receive_and_check_complete_zlib_package(b"some")
 
         assert exc_info.value is mock_exception
-        transport_impl.receive.assert_awaited_with(10)
-        transport_impl._handle_other_message.assert_called_once_with(response)
+        transport_impl._ws.receive.assert_awaited_with()
+        handle_other_message.assert_called_once_with(response)
 
     @pytest.mark.asyncio()
     async def test__receive_and_check_complete_zlib_package(self, transport_impl):
         response1 = StubResponse(type=aiohttp.WSMsgType.BINARY, data=b"more")
         response2 = StubResponse(type=aiohttp.WSMsgType.BINARY, data=b"data")
         response3 = StubResponse(type=aiohttp.WSMsgType.BINARY, data=b"\x00\x00\xff\xff")
-        transport_impl.receive = mock.AsyncMock(side_effect=[response1, response2, response3])
-        transport_impl.zlib = mock.Mock(decompress=mock.Mock(return_value=b"decoded utf-8 encoded bytes"))
+        transport_impl._ws.receive = mock.AsyncMock(side_effect=[response1, response2, response3])
+        transport_impl._zlib = mock.Mock(decompress=mock.Mock(return_value=b"decoded utf-8 encoded bytes"))
 
-        assert (
-            await transport_impl._receive_and_check_complete_zlib_package(b"some", 10) == "decoded utf-8 encoded bytes"
-        )
+        assert await transport_impl._receive_and_check_complete_zlib_package(b"some") == "decoded utf-8 encoded bytes"
 
-        assert transport_impl.receive.call_count == 3
-        transport_impl.receive.assert_has_awaits([mock.call(10), mock.call(10), mock.call(10)])
-        transport_impl.zlib.decompress.assert_called_once_with(bytearray(b"somemoredata\x00\x00\xff\xff"))
+        assert transport_impl._ws.receive.call_count == 3
+        transport_impl._ws.receive.assert_has_awaits([mock.call(), mock.call(), mock.call()])
+        transport_impl._zlib.decompress.assert_called_once_with(bytearray(b"somemoredata\x00\x00\xff\xff"))
 
+    @pytest.mark.parametrize("transport_compression", [True, False])
     @pytest.mark.asyncio()
-    async def test_connect_yields_websocket(self, http_settings, proxy_settings):
-        class MockWS(hikari_test_helpers.AsyncContextManagerMock, shard._GatewayTransport):
-            closed = True
-            send_close = mock.AsyncMock()
-            sent_close = False
-
-            def __init__(self):
-                pass
-
-        mock_websocket = MockWS()
-        mock_client_session = hikari_test_helpers.AsyncContextManagerMock()
-        mock_client_session.ws_connect = mock.MagicMock(return_value=mock_websocket)
+    async def test_connect(self, http_settings, proxy_settings, transport_compression):
+        logger = mock.Mock()
+        log_filterer = mock.Mock()
+        client_session = mock.Mock()
+        websocket = mock.Mock()
+        exit_stack = mock.AsyncMock(enter_async_context=mock.AsyncMock(side_effect=[client_session, websocket]))
 
         stack = contextlib.ExitStack()
         sleep = stack.enter_context(mock.patch.object(asyncio, "sleep"))
-        client_session = stack.enter_context(
-            mock.patch.object(aiohttp, "ClientSession", return_value=mock_client_session)
-        )
-        tcp_connector = stack.enter_context(mock.patch.object(aiohttp, "TCPConnector"))
-        client_timeout = stack.enter_context(mock.patch.object(aiohttp, "ClientTimeout"))
-        logger = mock.Mock()
-        log_filterer = mock.Mock()
+        create_tcp_connector = stack.enter_context(mock.patch.object(net, "create_tcp_connector"))
+        create_client_session = stack.enter_context(mock.patch.object(net, "create_client_session"))
+        stack.enter_context(mock.patch.object(contextlib, "AsyncExitStack", return_value=exit_stack))
 
         with stack:
-            async with shard._GatewayTransport.connect(
+            ws = await shard._GatewayTransport.connect(
                 http_settings=http_settings,
                 proxy_settings=proxy_settings,
                 logger=logger,
-                url="https://some.url",
+                url="testing.com",
                 log_filterer=log_filterer,
-            ) as ws:
-                assert ws.logger is logger
-
-        tcp_connector.assert_called_once_with(
-            limit=1,
-            ttl_dns_cache=10,
-            use_dns_cache=False,
-            ssl=http_settings.ssl,
-            enable_cleanup_closed=http_settings.enable_cleanup_closed,
-            force_close=http_settings.force_close_transports,
-        )
-        client_timeout.assert_called_once_with(
-            total=http_settings.timeouts.total,
-            connect=http_settings.timeouts.acquire_and_connect,
-            sock_read=http_settings.timeouts.request_socket_read,
-            sock_connect=http_settings.timeouts.request_socket_connect,
+                transport_compression=transport_compression,
+            )
+
+        assert isinstance(ws, shard._GatewayTransport)
+        assert ws._ws is websocket
+        assert ws._exit_stack is exit_stack
+        assert ws._logger is logger
+        assert ws._log_filterer is log_filterer
+
+        if transport_compression:
+            assert ws._receive_and_check == ws._receive_and_check_zlib
+        else:
+            assert ws._receive_and_check == ws._receive_and_check_text
+
+        assert exit_stack.enter_async_context.call_count == 2
+        exit_stack.enter_async_context.assert_has_calls(
+            [mock.call(create_client_session.return_value), mock.call(client_session.ws_connect.return_value)]
         )
-        client_session.assert_called_once_with(
-            connector=tcp_connector(),
+
+        create_tcp_connector.assert_called_once_with(http_settings=http_settings, dns_cache=False, limit=1)
+        create_client_session.assert_called_once_with(
+            connector=create_tcp_connector.return_value,
             connector_owner=True,
             raise_for_status=True,
-            timeout=client_timeout(),
+            http_settings=http_settings,
             trust_env=proxy_settings.trust_env,
-            version=aiohttp.HttpVersion11,
-            ws_response_class=shard._GatewayTransport,
         )
-        mock_client_session.ws_connect.assert_called_once_with(
+        client_session.ws_connect.assert_called_once_with(
             max_msg_size=0,
             proxy=proxy_settings.url,
             proxy_headers=proxy_settings.headers,
-            url="https://some.url",
+            url="testing.com",
             autoclose=False,
         )
-        mock_client_session.assert_used_once()
-        mock_websocket.assert_used_once()
-        sleep.assert_awaited_once_with(0.25)
-
-    @pytest.mark.asyncio()
-    async def test_connect_when_gateway_error_after_connecting(self, http_settings, proxy_settings):
-        class MockWS(hikari_test_helpers.AsyncContextManagerMock, shard._GatewayTransport):
-            closed = False
-            sent_close = False
-            send_close = mock.AsyncMock()
-
-            def __init__(self):
-                pass
-
-        mock_websocket = MockWS()
-        mock_client_session = hikari_test_helpers.AsyncContextManagerMock()
-        mock_client_session.ws_connect = mock.MagicMock(return_value=mock_websocket)
-
-        stack = contextlib.ExitStack()
-        sleep = stack.enter_context(mock.patch.object(asyncio, "sleep"))
-        stack.enter_context(mock.patch.object(aiohttp, "ClientSession", return_value=mock_client_session))
-        stack.enter_context(mock.patch.object(aiohttp, "TCPConnector"))
-        stack.enter_context(mock.patch.object(aiohttp, "ClientTimeout"))
-        stack.enter_context(pytest.raises(errors.GatewayError, match="some reason"))
-        logger = mock.Mock()
-        log_filterer = mock.Mock()
-
-        with stack:
-            async with shard._GatewayTransport.connect(
-                http_settings=http_settings,
-                proxy_settings=proxy_settings,
-                logger=logger,
-                url="https://some.url",
-                log_filterer=log_filterer,
-            ):
-                raise errors.GatewayError("some reason")
-
-        mock_websocket.send_close.assert_awaited_once_with(
-            code=errors.ShardCloseCode.UNEXPECTED_CONDITION, message=b"unexpected fatal client error :-("
-        )
-
-        sleep.assert_awaited_once_with(0.25)
-        mock_client_session.assert_used_once()
-        mock_websocket.assert_used_once()
-
-    @pytest.mark.asyncio()
-    async def test_connect_when_unexpected_error_after_connecting(self, http_settings, proxy_settings):
-        class MockWS(hikari_test_helpers.AsyncContextManagerMock, shard._GatewayTransport):
-            closed = False
-            send_close = mock.AsyncMock()
-            sent_close = False
-
-            def __init__(self):
-                pass
-
-        mock_websocket = MockWS()
-        mock_client_session = hikari_test_helpers.AsyncContextManagerMock()
-        mock_client_session.ws_connect = mock.MagicMock(return_value=mock_websocket)
-
-        stack = contextlib.ExitStack()
-        sleep = stack.enter_context(mock.patch.object(asyncio, "sleep"))
-        stack.enter_context(mock.patch.object(aiohttp, "ClientSession", return_value=mock_client_session))
-        stack.enter_context(mock.patch.object(aiohttp, "TCPConnector"))
-        stack.enter_context(mock.patch.object(aiohttp, "ClientTimeout"))
-        stack.enter_context(pytest.raises(errors.GatewayError, match="Unexpected ValueError: testing"))
-        logger = mock.Mock()
-        log_filterer = mock.Mock()
-
-        with stack:
-            async with shard._GatewayTransport.connect(
-                http_settings=http_settings,
-                proxy_settings=proxy_settings,
-                logger=logger,
-                url="https://some.url",
-                log_filterer=log_filterer,
-            ):
-                raise ValueError("testing")
-
-        mock_websocket.send_close.assert_awaited_once_with(
-            code=errors.ShardCloseCode.UNEXPECTED_CONDITION, message=b"unexpected fatal client error :-("
-        )
-
-        sleep.assert_awaited_once_with(0.25)
-        mock_client_session.assert_used_once()
-        mock_websocket.assert_used_once()
+        exit_stack.aclose.assert_not_called()
+        sleep.assert_not_called()
 
     @pytest.mark.asyncio()
-    async def test_connect_when_no_error_and_not_closing(self, http_settings, proxy_settings):
-        class MockWS(hikari_test_helpers.AsyncContextManagerMock, shard._GatewayTransport):
-            closed = False
-            _closing = False
-            sent_close = False
-            send_close = mock.AsyncMock()
-
-            def __init__(self):
-                pass
-
-        mock_websocket = MockWS()
-        mock_client_session = hikari_test_helpers.AsyncContextManagerMock()
-        mock_client_session.ws_connect = mock.MagicMock(return_value=mock_websocket)
-
-        stack = contextlib.ExitStack()
-        sleep = stack.enter_context(mock.patch.object(asyncio, "sleep"))
-        stack.enter_context(mock.patch.object(aiohttp, "ClientSession", return_value=mock_client_session))
-        stack.enter_context(mock.patch.object(aiohttp, "TCPConnector"))
-        stack.enter_context(mock.patch.object(aiohttp, "ClientTimeout"))
+    async def test_connect_when_error_while_connecting(self, http_settings, proxy_settings):
         logger = mock.Mock()
         log_filterer = mock.Mock()
-
-        with stack:
-            async with shard._GatewayTransport.connect(
-                http_settings=http_settings,
-                proxy_settings=proxy_settings,
-                logger=logger,
-                url="https://some.url",
-                log_filterer=log_filterer,
-            ):
-                pass
-
-        mock_websocket.send_close.assert_awaited_once_with(
-            code=shard._RESUME_CLOSE_CODE, message=b"client is shutting down"
-        )
-
-        sleep.assert_awaited_once_with(0.25)
-        mock_client_session.assert_used_once()
-        mock_websocket.assert_used_once()
-
-    @pytest.mark.asyncio()
-    async def test_connect_when_no_error_and_closing(self, http_settings, proxy_settings):
-        class MockWS(hikari_test_helpers.AsyncContextManagerMock, shard._GatewayTransport):
-            closed = False
-            _closing = True
-            close = mock.AsyncMock()
-
-            def __init__(self):
-                pass
-
-        mock_websocket = MockWS()
-        mock_client_session = hikari_test_helpers.AsyncContextManagerMock()
-        mock_client_session.ws_connect = mock.MagicMock(return_value=mock_websocket)
+        client_session = mock.Mock()
+        websocket = mock.Mock()
+        exit_stack = mock.AsyncMock(enter_async_context=mock.AsyncMock(side_effect=[client_session, websocket]))
 
         stack = contextlib.ExitStack()
         sleep = stack.enter_context(mock.patch.object(asyncio, "sleep"))
-        stack.enter_context(mock.patch.object(aiohttp, "ClientSession", return_value=mock_client_session))
-        stack.enter_context(mock.patch.object(aiohttp, "TCPConnector"))
-        stack.enter_context(mock.patch.object(aiohttp, "ClientTimeout"))
-        logger = mock.Mock()
-        log_filterer = mock.Mock()
+        stack.enter_context(mock.patch.object(net, "create_tcp_connector", side_effect=RuntimeError))
+        stack.enter_context(mock.patch.object(contextlib, "AsyncExitStack", return_value=exit_stack))
+        stack.enter_context(pytest.raises(RuntimeError))
 
         with stack:
-            async with shard._GatewayTransport.connect(
+            await shard._GatewayTransport.connect(
                 http_settings=http_settings,
                 proxy_settings=proxy_settings,
                 logger=logger,
                 url="https://some.url",
                 log_filterer=log_filterer,
-            ):
-                pass
-
-        mock_websocket.close.assert_not_called()
+                transport_compression=True,
+            )
 
+        exit_stack.aclose.assert_awaited_once_with()
         sleep.assert_awaited_once_with(0.25)
-        mock_client_session.assert_used_once()
-        mock_websocket.assert_used_once()
 
     @pytest.mark.asyncio()
     @pytest.mark.parametrize(
@@ -529,77 +448,43 @@ def __init__(self):
                 aiohttp.WSServerHandshakeError(status=123, message="some error", request_info=None, history=None),
                 "WSServerHandshakeError(None, None, status=123, message='some error')",
             ),
-            (aiohttp.ClientOSError("some error"), "some error"),
+            (aiohttp.ClientOSError("some os error"), "some os error"),
             (aiohttp.ClientConnectionError("some error"), "some error"),
             (asyncio.TimeoutError("some error"), "Timeout exceeded"),
         ],
     )
-    async def test_connect_when_expected_error_connecting(self, http_settings, proxy_settings, error, reason):
-        mock_client_session = hikari_test_helpers.AsyncContextManagerMock()
-        mock_client_session.ws_connect = mock.MagicMock(side_effect=error)
+    async def test_connect_when_expected_error_while_connecting(self, http_settings, proxy_settings, error, reason):
+        logger = mock.Mock()
+        log_filterer = mock.Mock()
+        client_session = mock.Mock()
+        websocket = mock.Mock()
+        exit_stack = mock.AsyncMock(enter_async_context=mock.AsyncMock(side_effect=[client_session, websocket]))
 
         stack = contextlib.ExitStack()
         sleep = stack.enter_context(mock.patch.object(asyncio, "sleep"))
-        stack.enter_context(mock.patch.object(aiohttp, "ClientSession", return_value=mock_client_session))
-        stack.enter_context(mock.patch.object(aiohttp, "TCPConnector"))
-        stack.enter_context(mock.patch.object(aiohttp, "ClientTimeout"))
+        stack.enter_context(mock.patch.object(net, "create_tcp_connector", side_effect=error))
+        stack.enter_context(mock.patch.object(contextlib, "AsyncExitStack", return_value=exit_stack))
         stack.enter_context(
             pytest.raises(errors.GatewayConnectionError, match=re.escape(f"Failed to connect to server: {reason!r}"))
         )
-        logger = mock.Mock()
-        log_filterer = mock.Mock()
-
-        with stack:
-            async with shard._GatewayTransport.connect(
-                http_settings=http_settings,
-                proxy_settings=proxy_settings,
-                logger=logger,
-                url="https://some.url",
-                log_filterer=log_filterer,
-            ):
-                pass
-
-        sleep.assert_awaited_once_with(0.25)
-        mock_client_session.assert_used_once()
-
-    @pytest.mark.asyncio()
-    async def test_connect_when_unexpected_error_connecting(self, http_settings, proxy_settings):
-        mock_client_session = hikari_test_helpers.AsyncContextManagerMock()
-        mock_client_session.ws_connect = mock.MagicMock(side_effect=RuntimeError("in tests"))
-
-        stack = contextlib.ExitStack()
-        sleep = stack.enter_context(mock.patch.object(asyncio, "sleep"))
-        stack.enter_context(mock.patch.object(aiohttp, "ClientSession", return_value=mock_client_session))
-        stack.enter_context(mock.patch.object(aiohttp, "TCPConnector"))
-        stack.enter_context(mock.patch.object(aiohttp, "ClientTimeout"))
-        stack.enter_context(pytest.raises(RuntimeError, match="in tests"))
-        logger = mock.Mock()
-        log_filterer = mock.Mock()
 
         with stack:
-            async with shard._GatewayTransport.connect(
+            await shard._GatewayTransport.connect(
                 http_settings=http_settings,
                 proxy_settings=proxy_settings,
                 logger=logger,
                 url="https://some.url",
                 log_filterer=log_filterer,
-            ):
-                pass
+                transport_compression=True,
+            )
 
+        exit_stack.aclose.assert_awaited_once_with()
         sleep.assert_awaited_once_with(0.25)
-        mock_client_session.assert_used_once()
-
-
-@pytest.fixture()
-def client_session():
-    stub = client_session_stub.ClientSessionStub()
-    with mock.patch.object(aiohttp, "ClientSession", new=stub):
-        yield stub
 
 
 @pytest.fixture()
 def client(http_settings, proxy_settings):
-    return hikari_test_helpers.mock_class_namespace(shard.GatewayShardImpl, slots_=False)(
+    return shard.GatewayShardImpl(
         event_manager=mock.Mock(),
         event_factory=mock.Mock(),
         url="wss://gateway.discord.gg",
@@ -611,6 +496,20 @@ def client(http_settings, proxy_settings):
 
 
 class TestGatewayShardImpl:
+    def test__init__when_unsupported_compression_format(self):
+        with pytest.raises(NotImplementedError, match=r"Unsupported compression format something"):
+            shard.GatewayShardImpl(
+                event_manager=mock.Mock(),
+                event_factory=mock.Mock(),
+                http_settings=http_settings,
+                proxy_settings=proxy_settings,
+                intents=intents.Intents.ALL,
+                url="wss://gaytewhuy.discord.meh",
+                data_format="json",
+                compression="something",
+                token="12345",
+            )
+
     def test_using_etf_is_unsupported(self, http_settings, proxy_settings):
         with pytest.raises(NotImplementedError, match="Unsupported gateway data format: etf"):
             shard.GatewayShardImpl(
@@ -639,115 +538,45 @@ def test_intents_property(self, client):
         assert client.intents is mock_intents
 
     @pytest.mark.parametrize(
-        ("ws", "expected"), [(None, False), (mock.Mock(sent_close=True), False), (mock.Mock(sent_close=False), True)]
+        ("keep_alive_task", "expected"),
+        [(None, False), ("some", True)],
     )
-    def test_is_alive_property(self, client, ws, expected):
-        client._ws = ws
+    def test_is_alive_property(self, client, keep_alive_task, expected):
+        client._keep_alive_task = keep_alive_task
+
         assert client.is_alive is expected
 
+    @pytest.mark.parametrize(
+        ("ws", "handshake_event", "expected"),
+        [
+            (None, None, False),
+            (None, True, False),
+            (None, False, False),
+            ("something", None, False),
+            ("something", True, True),
+            ("something", False, False),
+        ],
+    )
+    def test_is_connected_property(self, client, ws, handshake_event, expected):
+        client._ws = ws
+        client._handshake_event = (
+            None if handshake_event is None else mock.Mock(is_set=mock.Mock(return_value=handshake_event))
+        )
+
+        assert client.is_connected is expected
+
     def test_shard_count_property(self, client):
         client._shard_count = 69
         assert client.shard_count == 69
 
-    def test_shard__check_if_alive_when_not_alive(self, client):
-        with mock.patch.object(shard.GatewayShardImpl, "is_alive", new=False):
+    def test_shard__check_if_connected_when_not_alive(self, client):
+        with mock.patch.object(shard.GatewayShardImpl, "is_connected", new=False):
             with pytest.raises(errors.ComponentStateConflictError):
-                client._check_if_alive()
-
-    def test_shard__check_if_alive_when_alive(self, client):
-        with mock.patch.object(shard.GatewayShardImpl, "is_alive", new=True):
-            client._check_if_alive()
-
-    def test__get_ws_when_active(self, client):
-        mock_ws = client._ws = object()
+                client._check_if_connected()
 
-        assert client._get_ws() is mock_ws
-
-    def test__get_ws_when_inactive(self, client):
-        client._ws = None
-
-        with pytest.raises(errors.ComponentStateConflictError):
-            client._get_ws()
-
-    def test_dispatch_when_READY(self, client):
-        client._seq = 0
-        client._resume_gateway_url = None
-        client._session_id = 0
-        client._user_id = 0
-        client._logger = mock.Mock()
-        client._handshake_completed = mock.Mock()
-        client._event_manager = mock.Mock()
-
-        pl = {
-            "session_id": 101,
-            "resume_gateway_url": "testing.com",
-            "user": {"id": 123, "username": "hikari", "discriminator": "5863"},
-            "guilds": [
-                {"id": "123"},
-                {"id": "456"},
-                {"id": "789"},
-            ],
-            "v": 8,
-        }
-
-        client._dispatch(
-            "READY",
-            10,
-            pl,
-        )
-
-        assert client._seq == 10
-        assert client._resume_gateway_url == "testing.com"
-        assert client._session_id == 101
-        assert client._user_id == 123
-        client._logger.info.assert_called_once_with(
-            "shard is ready: %s guilds, %s (%s), session %r on v%s gateway",
-            3,
-            "hikari#5863",
-            123,
-            101,
-            8,
-        )
-        client._handshake_completed.set.assert_called_once_with()
-        client._event_manager.consume_raw_event.assert_called_once_with(
-            "READY",
-            client,
-            pl,
-        )
-
-    def test__dipatch_when_RESUMED(self, client):
-        client._seq = 0
-        client._session_id = 123
-        client._logger = mock.Mock()
-        client._handshake_completed = mock.Mock()
-        client._event_manager = mock.Mock()
-
-        client._dispatch("RESUMED", 10, {})
-
-        assert client._seq == 10
-        client._logger.info.assert_called_once_with("shard has resumed [session:%s, seq:%s]", 123, 10)
-        client._handshake_completed.set.assert_called_once_with()
-        client._event_manager.consume_raw_event.assert_called_once_with("RESUMED", client, {})
-
-    def test__dipatch(self, client):
-        client._logger = mock.Mock()
-        client._handshake_completed = mock.Mock()
-        client._event_manager = mock.Mock()
-
-        client._dispatch("EVENT NAME", 10, {"payload": None})
-
-        client._logger.info.assert_not_called()
-        client._logger.debug.assert_not_called()
-        client._handshake_completed.set.assert_not_called()
-        client._event_manager.consume_raw_event.assert_called_once_with("EVENT NAME", client, {"payload": None})
-
-    def test__serialize_activity_when_activity_is_None(self, client):
-        assert client._serialize_activity(None) is None
-
-    def test__serialize_activity_when_activity_is_not_None(self, client):
-        activity = mock.Mock(type="0", url="https://some.url")
-        activity.name = "some name"  # This has to be set separate because if not, its set as the mock's name
-        assert client._serialize_activity(activity) == {"name": "some name", "type": 0, "url": "https://some.url"}
+    def test_shard__check_if_connected_when_alive(self, client):
+        with mock.patch.object(shard.GatewayShardImpl, "is_connected", new=True):
+            client._check_if_connected()
 
     @pytest.mark.parametrize("idle_since", [datetime.datetime.now(), None])
     @pytest.mark.parametrize("afk", [True, False])
@@ -766,7 +595,7 @@ def test__serialize_and_store_presence_payload_when_all_args_undefined(
 
         actual_result = client._serialize_and_store_presence_payload()
 
-        if activity is not undefined.UNDEFINED and activity is not None:
+        if activity is not None:
             expected_activity = {
                 "name": activity.name,
                 "type": activity.type,
@@ -783,7 +612,7 @@ def test__serialize_and_store_presence_payload_when_all_args_undefined(
         expected_result = {
             "game": expected_activity,
             "since": int(idle_since.timestamp() * 1_000) if idle_since is not None else None,
-            "afk": afk if afk is not undefined.UNDEFINED else False,
+            "afk": afk,
             "status": expected_status,
         }
 
@@ -804,363 +633,681 @@ def test__serialize_and_store_presence_payload_sets_state(self, client, idle_sin
         assert client._is_afk == afk
         assert client._status == status
 
-    def test__serialize_datetime_when_datetime_is_None(self, client):
-        assert client._serialize_datetime(None) is None
+    def test_get_user_id(self, client):
+        client._user_id = 123
+
+        with mock.patch.object(shard.GatewayShardImpl, "_check_if_connected") as check_if_alive:
+            assert client.get_user_id() == 123
 
-    def test__serialize_datetime_when_datetime_is_not_None(self, client):
-        dt = datetime.datetime(2004, 11, 22, tzinfo=datetime.timezone.utc)
-        assert client._serialize_datetime(dt) == 1101081600000
+        check_if_alive.assert_called_once_with()
 
 
 @pytest.mark.asyncio()
 class TestGatewayShardImplAsync:
+    async def test_close_when_no_keep_alive_task(self, client):
+        client._keep_alive_task = None
+
+        with pytest.raises(errors.ComponentStateConflictError):
+            await client.close()
+
     async def test_close_when_closing_event_set(self, client):
-        client._check_if_alive = mock.Mock()
-        client._closing_event = mock.Mock(is_set=mock.Mock(return_value=True))
-        client._closed_event = mock.Mock(wait=mock.AsyncMock())
-        client._send_close = mock.Mock()
-        client._chunking_rate_limit = mock.Mock()
+        client._keep_alive_task = mock.Mock(cancel=mock.AsyncMock())
+        client._non_priority_rate_limit = mock.Mock()
         client._total_rate_limit = mock.Mock()
+        client._is_closing = True
 
-        await client.close()
+        with mock.patch.object(shard.GatewayShardImpl, "join") as join:
+            await client.close()
 
-        client._closing_event.set.assert_not_called()
-        client._send_close.assert_not_called()
-        client._chunking_rate_limit.close.assert_not_called()
+        join.assert_awaited_once_with()
+        client._keep_alive_task.cancel.assert_not_called()
+        client._non_priority_rate_limit.close.assert_not_called()
         client._total_rate_limit.close.assert_not_called()
-        client._closed_event.wait.assert_awaited_once_with()
 
     async def test_close_when_closing_event_not_set(self, client):
-        client._check_if_alive = mock.Mock()
-        client._closing_event = mock.Mock(is_set=mock.Mock(return_value=False))
-        client._closed_event = mock.Mock(wait=mock.AsyncMock())
-        client._ws = mock.Mock(send_close=mock.AsyncMock())
-        client._chunking_rate_limit = mock.Mock()
-        client._total_rate_limit = mock.Mock()
+        cancel_async_mock = mock.Mock()
 
-        await client.close()
+        class TaskMock:
+            def __init__(self):
+                self._awaited_count = 0
 
-        client._closing_event.set.assert_called_once_with()
-        client._ws.send_close.assert_awaited_once_with(
-            code=errors.ShardCloseCode.GOING_AWAY, message=b"shard disconnecting"
-        )
-        client._chunking_rate_limit.close.assert_called_once_with()
-        client._total_rate_limit.close.assert_called_once_with()
-        client._closed_event.wait.assert_awaited_once_with()
-
-    async def test_close_when_closing_event_not_set_and_ws_is_None(self, client):
-        client._check_if_alive = mock.Mock()
-        client._closing_event = mock.Mock(is_set=mock.Mock(return_value=False))
-        client._closed_event = mock.Mock(wait=mock.AsyncMock())
-        client._ws = None
-        client._chunking_rate_limit = mock.Mock()
+            def __await__(self):
+                self._awaited_count += 1
+                raise asyncio.CancelledError
+
+            @property
+            def cancel(self):
+                return cancel_async_mock
+
+            def assert_awaited_once(self):
+                assert self._awaited_count == 1
+
+        client._keep_alive_task = keep_alive_task = TaskMock()
+        client._non_priority_rate_limit = mock.Mock()
         client._total_rate_limit = mock.Mock()
 
-        await client.close()
+        with mock.patch.object(shard.GatewayShardImpl, "join") as join:
+            await client.close()
 
-        client._closing_event.set.assert_called_once_with()
-        client._chunking_rate_limit.close.assert_called_once_with()
+        join.assert_not_called()
+        cancel_async_mock.assert_called_once_with()
+        keep_alive_task.assert_awaited_once()
+        client._non_priority_rate_limit.close.assert_called_once_with()
         client._total_rate_limit.close.assert_called_once_with()
-        client._closed_event.wait.assert_awaited_once_with()
 
-    async def test_when__user_id_is_None(self, client):
-        client._handshake_completed = mock.Mock(wait=mock.AsyncMock())
-        client._user_id = None
-        with pytest.raises(RuntimeError):
-            assert await client.get_user_id()
+    async def test_join_when_not_alive(self, client):
+        client._keep_alive_task = None
 
-    async def test_when__user_id_is_not_None(self, client):
-        client._handshake_completed = mock.Mock(wait=mock.AsyncMock())
-        client._user_id = 123
-        assert await client.get_user_id() == 123
+        with pytest.raises(errors.ComponentStateConflictError):
+            await client.join()
 
     async def test_join(self, client):
-        client._check_if_alive = mock.Mock()
-        client._closed_event = mock.Mock(wait=mock.AsyncMock())
+        client._keep_alive_task = object()
+
+        with mock.patch.object(asyncio, "wait_for") as wait_for:
+            with mock.patch.object(asyncio, "shield", new=mock.Mock()) as shield:
+                await client.join()
+
+        shield.assert_called_once_with(client._keep_alive_task)
+        wait_for.assert_awaited_once_with(shield.return_value, timeout=None)
+
+    async def test__send_json(self, client):
+        client._total_rate_limit = mock.AsyncMock()
+        client._non_priority_rate_limit = mock.AsyncMock()
+        client._ws = mock.AsyncMock()
+        data = object()
+
+        await client._send_json(data)
+
+        client._non_priority_rate_limit.acquire.assert_awaited_once_with()
+        client._total_rate_limit.acquire.assert_awaited_once_with()
+        client._ws.send_json.assert_awaited_once_with(data)
+
+    async def test__send_json_when_priority(self, client):
+        client._total_rate_limit = mock.AsyncMock()
+        client._non_priority_rate_limit = mock.AsyncMock()
+        client._ws = mock.AsyncMock()
+        data = object()
 
-        await client.join()
+        await client._send_json(data, priority=True)
 
-        client._closed_event.wait.assert_awaited_once_with()
+        client._non_priority_rate_limit.acquire.assert_not_called()
+        client._total_rate_limit.acquire.assert_awaited_once_with()
+        client._ws.send_json.assert_awaited_once_with(data)
 
     async def test_request_guild_members_when_no_query_and_no_limit_and_GUILD_MEMBERS_not_enabled(self, client):
-        client._check_if_alive = mock.Mock()
         client._intents = intents.Intents.GUILD_INTEGRATIONS
 
-        with pytest.raises(errors.MissingIntentError):
-            await client.request_guild_members(123, query="", limit=0)
-        client._check_if_alive.assert_called_once_with()
+        with mock.patch.object(shard.GatewayShardImpl, "_check_if_connected") as check_if_alive:
+            with pytest.raises(errors.MissingIntentError):
+                await client.request_guild_members(123, query="", limit=0)
+
+        check_if_alive.assert_called_once_with()
 
     async def test_request_guild_members_when_presences_and_GUILD_PRESENCES_not_enabled(self, client):
-        client._check_if_alive = mock.Mock()
         client._intents = intents.Intents.GUILD_INTEGRATIONS
 
-        with pytest.raises(errors.MissingIntentError):
-            await client.request_guild_members(123, query="test", limit=1, include_presences=True)
-        client._check_if_alive.assert_called_once_with()
+        with mock.patch.object(shard.GatewayShardImpl, "_check_if_connected") as check_if_alive:
+            with pytest.raises(errors.MissingIntentError):
+                await client.request_guild_members(123, query="test", limit=1, include_presences=True)
+
+        check_if_alive.assert_called_once_with()
 
     async def test_request_guild_members_when_presences_false_and_GUILD_PRESENCES_not_enabled(self, client):
-        client._check_if_alive = mock.Mock()
         client._intents = intents.Intents.GUILD_INTEGRATIONS
-        client._send_json = mock.AsyncMock()
 
-        await client.request_guild_members(123, query="test", limit=1, include_presences=False)
+        with mock.patch.object(shard.GatewayShardImpl, "_send_json") as send_json:
+            with mock.patch.object(shard.GatewayShardImpl, "_check_if_connected") as check_if_alive:
+                await client.request_guild_members(123, query="test", limit=1, include_presences=False)
 
-        client._send_json.assert_awaited_once_with(
+        send_json.assert_awaited_once_with(
             {
                 "op": 8,
                 "d": {"guild_id": "123", "query": "test", "presences": False, "limit": 1},
             }
         )
-        client._check_if_alive.assert_called_once_with()
+
+        check_if_alive.assert_called_once_with()
 
     @pytest.mark.parametrize("kwargs", [{"query": "some query"}, {"limit": 1}])
     async def test_request_guild_members_when_specifiying_users_with_limit_or_query(self, client, kwargs):
-        client._check_if_alive = mock.Mock()
         client._intents = intents.Intents.GUILD_INTEGRATIONS
 
-        with pytest.raises(ValueError, match="Cannot specify limit/query with users"):
-            await client.request_guild_members(123, users=[], **kwargs)
-        client._check_if_alive.assert_called_once_with()
+        with mock.patch.object(shard.GatewayShardImpl, "_check_if_connected") as check_if_alive:
+            with pytest.raises(ValueError, match="Cannot specify limit/query with users"):
+                await client.request_guild_members(123, users=[], **kwargs)
+
+        check_if_alive.assert_called_once_with()
 
     @pytest.mark.parametrize("limit", [-1, 101])
     async def test_request_guild_members_when_limit_under_0_or_over_100(self, client, limit):
-        client._check_if_alive = mock.Mock()
         client._intents = intents.Intents.ALL
 
-        with pytest.raises(ValueError, match="'limit' must be between 0 and 100, both inclusive"):
-            await client.request_guild_members(123, limit=limit)
-        client._check_if_alive.assert_called_once_with()
+        with mock.patch.object(shard.GatewayShardImpl, "_check_if_connected") as check_if_alive:
+            with pytest.raises(ValueError, match="'limit' must be between 0 and 100, both inclusive"):
+                await client.request_guild_members(123, limit=limit)
+
+        check_if_alive.assert_called_once_with()
 
     async def test_request_guild_members_when_users_over_100(self, client):
-        client._check_if_alive = mock.Mock()
         client._intents = intents.Intents.ALL
 
-        with pytest.raises(ValueError, match="'users' is limited to 100 users"):
-            await client.request_guild_members(123, users=range(101))
-        client._check_if_alive.assert_called_once_with()
+        with mock.patch.object(shard.GatewayShardImpl, "_check_if_connected") as check_if_alive:
+            with pytest.raises(ValueError, match="'users' is limited to 100 users"):
+                await client.request_guild_members(123, users=range(101))
+
+        check_if_alive.assert_called_once_with()
 
     async def test_request_guild_members_when_nonce_over_32_chars(self, client):
-        client._check_if_alive = mock.Mock()
         client._intents = intents.Intents.ALL
 
-        with pytest.raises(ValueError, match="'nonce' can be no longer than 32 byte characters long."):
-            await client.request_guild_members(123, nonce="x" * 33)
-        client._check_if_alive.assert_called_once_with()
+        with mock.patch.object(shard.GatewayShardImpl, "_check_if_connected") as check_if_alive:
+            with pytest.raises(ValueError, match="'nonce' can be no longer than 32 byte characters long."):
+                await client.request_guild_members(123, nonce="x" * 33)
+
+        check_if_alive.assert_called_once_with()
 
     @pytest.mark.parametrize("include_presences", [True, False])
     async def test_request_guild_members(self, client, include_presences):
         client._intents = intents.Intents.ALL
-        client._check_if_alive = mock.Mock()
-        client._send_json = mock.AsyncMock()
 
-        await client.request_guild_members(123, include_presences=include_presences)
+        with mock.patch.object(shard.GatewayShardImpl, "_send_json") as send_json:
+            with mock.patch.object(shard.GatewayShardImpl, "_check_if_connected") as check_if_alive:
+                await client.request_guild_members(123, include_presences=include_presences)
 
-        client._send_json.assert_awaited_once_with(
+        send_json.assert_awaited_once_with(
             {
                 "op": 8,
                 "d": {"guild_id": "123", "query": "", "presences": include_presences, "limit": 0},
             }
         )
-        client._check_if_alive.assert_called_once_with()
+        check_if_alive.assert_called_once_with()
 
-    async def test_start_when_already_running(self, client):
-        client._run_task = object()
+    @pytest.mark.parametrize("attr", ["_keep_alive_task", "_handshake_event"])
+    async def test_start_when_already_running(self, client, attr):
+        setattr(client, attr, object())
 
         with pytest.raises(errors.ComponentStateConflictError):
             await client.start()
 
     async def test_start_when_shard_closed_before_starting(self, client):
-        client._run_task = None
+        client._keep_alive_task = None
         client._shard_id = 20
-        client._run = mock.Mock()
-        client._handshake_completed = mock.Mock(wait=mock.Mock())
-        run_task = mock.Mock()
-        waiter = mock.Mock()
+        handshake_event = mock.Mock(is_set=mock.Mock(return_value=False))
 
         stack = contextlib.ExitStack()
-        create_task = stack.enter_context(mock.patch.object(asyncio, "create_task", side_effect=[run_task, waiter]))
-        wait = stack.enter_context(mock.patch.object(asyncio, "wait", return_value=([run_task], [waiter])))
-        stack.enter_context(
-            pytest.raises(asyncio.CancelledError, match="shard 20 was closed before it could start successfully")
-        )
+        stack.enter_context(mock.patch.object(aio, "first_completed"))
+        stack.enter_context(mock.patch.object(asyncio, "shield"))
+        stack.enter_context(mock.patch.object(asyncio, "create_task"))
+        stack.enter_context(mock.patch.object(shard.GatewayShardImpl, "_keep_alive", new=mock.Mock()))
+        stack.enter_context(mock.patch.object(asyncio, "Event", return_value=handshake_event))
+        stack.enter_context(pytest.raises(RuntimeError, match="shard 20 was closed before it could start successfully"))
 
         with stack:
             await client.start()
 
-        assert client._run_task is None
-
-        assert create_task.call_count == 2
-        create_task.has_call(mock.call(client._run(), name="run shard 20"))
-        create_task.has_call(mock.call(client._handshake_completed.wait(), name="wait for shard 20 to start"))
-
-        run_task.result.assert_called_once_with()
-        waiter.cancel.assert_called_once_with()
-        wait.assert_awaited_once_with((waiter, run_task), return_when=asyncio.FIRST_COMPLETED)
+        assert client._keep_alive_task is None
 
     async def test_start(self, client):
-        client._run_task = None
+        client._keep_alive_task = None
         client._shard_id = 20
-        client._run = mock.Mock()
-        client._handshake_completed = mock.Mock(wait=mock.Mock())
-        run_task = mock.Mock()
-        waiter = mock.Mock()
+        handshake_event = mock.Mock(is_set=mock.Mock(return_value=True))
+        keep_alive_task = mock.Mock()
 
-        with mock.patch.object(asyncio, "create_task", side_effect=[run_task, waiter]) as create_task:
-            with mock.patch.object(asyncio, "wait", return_value=([waiter], [run_task])) as wait:
-                await client.start()
+        stack = contextlib.ExitStack()
+        first_completed = stack.enter_context(mock.patch.object(aio, "first_completed"))
+        shield = stack.enter_context(mock.patch.object(asyncio, "shield"))
+        create_task = stack.enter_context(mock.patch.object(asyncio, "create_task", return_value=keep_alive_task))
+        keep_alive = stack.enter_context(mock.patch.object(shard.GatewayShardImpl, "_keep_alive", new=mock.Mock()))
+        stack.enter_context(mock.patch.object(asyncio, "Event", return_value=handshake_event))
 
-        assert client._run_task == run_task
+        with stack:
+            await client.start()
 
-        assert create_task.call_count == 2
-        create_task.has_call(mock.call(client._run(), name="run shard 20"))
-        create_task.has_call(mock.call(client._handshake_completed.wait(), name="wait for shard 20 to start"))
+        assert client._keep_alive_task is keep_alive_task
 
-        run_task.result.assert_not_called()
-        waiter.cancel.assert_called_once_with()
-        wait.assert_awaited_once_with((waiter, run_task), return_when=asyncio.FIRST_COMPLETED)
+        create_task.assert_called_once_with(keep_alive.return_value, name="keep alive (shard 20)")
+        shield.assert_called_once_with(create_task.return_value)
+        first_completed.assert_awaited_once_with(handshake_event.wait.return_value, shield.return_value)
 
     async def test_update_presence(self, client):
-        client._check_if_alive = mock.Mock()
-        presence_payload = object()
-        client._serialize_and_store_presence_payload = mock.Mock(return_value=presence_payload)
-        client._send_json = mock.AsyncMock()
-
-        await client.update_presence(
-            idle_since=datetime.datetime.now(),
-            afk=True,
-            status=presences.Status.IDLE,
-            activity=None,
+        with mock.patch.object(shard.GatewayShardImpl, "_serialize_and_store_presence_payload") as presence:
+            with mock.patch.object(shard.GatewayShardImpl, "_check_if_connected") as check_if_alive:
+                with mock.patch.object(shard.GatewayShardImpl, "_send_json") as send_json:
+                    await client.update_presence(
+                        idle_since=datetime.datetime.now(),
+                        afk=True,
+                        status=presences.Status.IDLE,
+                        activity=None,
+                    )
+
+        send_json.assert_awaited_once_with({"op": 3, "d": presence.return_value})
+        check_if_alive.assert_called_once_with()
+
+    async def test_update_voice_state(self, client):
+        with mock.patch.object(shard.GatewayShardImpl, "_check_if_connected") as check_if_alive:
+            with mock.patch.object(shard.GatewayShardImpl, "_send_json") as send_json:
+                await client.update_voice_state(123456, 6969420, self_mute=False, self_deaf=True)
+
+        send_json.assert_awaited_once_with(
+            {
+                "op": 4,
+                "d": {
+                    "guild_id": "123456",
+                    "channel_id": "6969420",
+                    "self_mute": False,
+                    "self_deaf": True,
+                },
+            }
         )
+        check_if_alive.assert_called_once_with()
 
-        client._send_json.assert_awaited_once_with({"op": 3, "d": presence_payload})
-        client._check_if_alive.assert_called_once_with()
+    async def test_update_voice_state_without_optionals(self, client):
+        with mock.patch.object(shard.GatewayShardImpl, "_check_if_connected") as check_if_alive:
+            with mock.patch.object(shard.GatewayShardImpl, "_send_json") as send_json:
+                await client.update_voice_state(123456, 6969420)
 
-    async def test_update_voice_state(self, client):
-        client._check_if_alive = mock.Mock()
-        client._send_json = mock.AsyncMock()
-        payload = {
-            "guild_id": "123456",
-            "channel_id": "6969420",
-            "self_mute": False,
-            "self_deaf": True,
-        }
+        send_json.assert_awaited_once_with({"op": 4, "d": {"guild_id": "123456", "channel_id": "6969420"}})
+        check_if_alive.assert_called_once_with()
 
-        await client.update_voice_state(123456, 6969420, self_mute=False, self_deaf=True)
+    @hikari_test_helpers.timeout()
+    async def test__heartbeat(self, client):
+        client._last_heartbeat_sent = 5
+        client._logger = mock.Mock()
 
-        client._send_json.assert_awaited_once_with({"op": 4, "d": payload})
+        class ExitException(Exception):
+            ...
 
-    async def test_update_voice_state_without_optionals(self, client):
-        client._check_if_alive = mock.Mock()
-        client._send_json = mock.AsyncMock()
-        payload = {"guild_id": "123456", "channel_id": "6969420"}
+        stack = contextlib.ExitStack()
+        sleep = stack.enter_context(mock.patch.object(asyncio, "sleep", side_effect=[None, ExitException]))
+        stack.enter_context(mock.patch.object(time, "monotonic", return_value=10))
+        send_heartbeat = stack.enter_context(mock.patch.object(shard.GatewayShardImpl, "_send_heartbeat"))
+        stack.enter_context(pytest.raises(ExitException))
 
-        await client.update_voice_state(123456, 6969420)
+        with stack:
+            await client._heartbeat(20)
 
-        client._send_json.assert_awaited_once_with({"op": 4, "d": payload})
+        assert send_heartbeat.await_count == 2
+        send_heartbeat.assert_has_awaits([mock.call(), mock.call()])
+        assert sleep.await_count == 2
+        sleep.assert_has_awaits([mock.call(20), mock.call(20)])
 
-    async def test__dispatch_for_unknown_event(self, client):
+    @hikari_test_helpers.timeout()
+    async def test__heartbeat_when_zombie(self, client):
+        client._last_heartbeat_sent = 10
         client._logger = mock.Mock()
-        client._handshake_completed = mock.Mock()
-        client._event_manager = mock.Mock(consume_raw_event=mock.Mock(side_effect=LookupError))
 
-        client._dispatch("UNEXISTING_EVENT", 10, {"payload": None})
+        with mock.patch.object(time, "monotonic", return_value=5):
+            with mock.patch.object(asyncio, "sleep") as sleep:
+                await client._heartbeat(20)
 
-        client._logger.info.assert_not_called()
-        client._handshake_completed.set.assert_not_called()
-        client._event_manager.consume_raw_event.assert_called_once_with("UNEXISTING_EVENT", client, {"payload": None})
-        client._logger.debug.assert_called_once_with(
-            "ignoring unknown event %s:\n    %r", "UNEXISTING_EVENT", {"payload": None}
-        )
+        sleep.assert_not_called()
+
+    async def test__connect_when_ws(self, client):
+        client._ws = object()
+
+        with pytest.raises(errors.ComponentStateConflictError):
+            await client._connect()
+
+    async def test__connect_when_not_reconnecting(self, client, http_settings, proxy_settings):
+        ws = mock.AsyncMock()
+        ws.receive_json.return_value = {"op": 10, "d": {"heartbeat_interval": 10}}
+        client._transport_compression = False
+        client._shard_id = 20
+        client._shard_count = 100
+        client._gateway_url = "wss://somewhere.com?somewhere=true"
+        client._resume_gateway_url = None
+        client._token = "sometoken"
+        client._logger = mock.Mock()
+        client._handshake_event = mock.Mock()
+        client._seq = None
+        client._large_threshold = "your mom"
+        client._intents = 9
+
+        heartbeat_task = object()
+        poll_events_task = object()
+        shielded_heartbeat_task = object()
+        shielded_poll_events_task = object()
 
-    async def test__identify(self, client):
-        client._token = "token"
-        client._intents = intents.Intents.ALL
-        client._large_threshold = 123
-        client._shard_id = 0
-        client._shard_count = 1
-        client._serialize_and_store_presence_payload = mock.Mock(return_value={"presence": "payload"})
-        client._send_json = mock.AsyncMock()
         stack = contextlib.ExitStack()
-        stack.enter_context(mock.patch.object(platform, "system", return_value="Potato PC"))
+        create_task = stack.enter_context(
+            mock.patch.object(asyncio, "create_task", side_effect=[heartbeat_task, poll_events_task])
+        )
+        shield = stack.enter_context(
+            mock.patch.object(asyncio, "shield", side_effect=[shielded_heartbeat_task, shielded_poll_events_task])
+        )
+        first_completed = stack.enter_context(mock.patch.object(aio, "first_completed"))
+        log_filterer = stack.enter_context(mock.patch.object(shard, "_log_filterer"))
+        serialize_and_store_presence_payload = stack.enter_context(
+            mock.patch.object(shard.GatewayShardImpl, "_serialize_and_store_presence_payload")
+        )
+        send_json = stack.enter_context(mock.patch.object(shard.GatewayShardImpl, "_send_json"))
+        heartbeat = stack.enter_context(mock.patch.object(shard.GatewayShardImpl, "_heartbeat", new=mock.Mock()))
+        poll_events = stack.enter_context(mock.patch.object(shard.GatewayShardImpl, "_poll_events", new=mock.Mock()))
+        gateway_transport_connect = stack.enter_context(
+            mock.patch.object(shard._GatewayTransport, "connect", return_value=ws)
+        )
+        stack.enter_context(mock.patch.object(shard, "_VERSION", new=400))
+        stack.enter_context(mock.patch.object(platform, "system", return_value="Potato OS"))
         stack.enter_context(mock.patch.object(platform, "architecture", return_value=["ARM64"]))
-        stack.enter_context(mock.patch.object(aiohttp, "__version__", new="v0.0.1"))
-        stack.enter_context(mock.patch.object(_about, "__version__", new="v1.0.0"))
+        stack.enter_context(mock.patch.object(aiohttp, "__version__", new="4.0"))
+        stack.enter_context(mock.patch.object(_about, "__version__", new="1.0.0"))
 
         with stack:
-            await client._identify()
-
-        expected_json = {
-            "op": 2,
-            "d": {
-                "token": "token",
-                "compress": False,
-                "large_threshold": 123,
-                "properties": {
-                    "os": "Potato PC ARM64",
-                    "browser": "hikari (v1.0.0, aiohttp v0.0.1)",
-                    "device": "hikari v1.0.0",
+            assert await client._connect() == (heartbeat_task, poll_events_task)
+
+        log_filterer.assert_called_once_with("sometoken")
+        gateway_transport_connect.assert_called_once_with(
+            http_settings=http_settings,
+            log_filterer=log_filterer.return_value,
+            logger=client._logger,
+            proxy_settings=proxy_settings,
+            transport_compression=False,
+            url="wss://somewhere.com?somewhere=true&v=400&encoding=json",
+        )
+        client._event_factory.deserialize_connected_event.assert_called_once_with(client)
+        client._event_manager.dispatch.assert_called_once_with(
+            client._event_factory.deserialize_connected_event.return_value
+        )
+
+        assert create_task.call_count == 2
+        create_task.assert_has_calls(
+            [
+                mock.call(heartbeat.return_value, name="heartbeat (shard 20)"),
+                mock.call(poll_events.return_value, name="poll events (shard 20)"),
+            ],
+        )
+        heartbeat.assert_called_once_with(0.01)
+
+        ws.receive_json.assert_awaited_once_with()
+        send_json.assert_called_once_with(
+            {
+                "op": 2,
+                "d": {
+                    "token": "sometoken",
+                    "compress": False,
+                    "large_threshold": "your mom",
+                    "properties": {
+                        "os": "Potato OS ARM64",
+                        "browser": "hikari (1.0.0, aiohttp 4.0)",
+                        "device": "hikari 1.0.0",
+                    },
+                    "shard": [20, 100],
+                    "intents": 9,
+                    "presence": serialize_and_store_presence_payload.return_value,
                 },
-                "shard": [0, 1],
-                "intents": 131071,
-                "presence": {"presence": "payload"},
-            },
-        }
-        client._send_json.assert_awaited_once_with(expected_json)
+            }
+        )
 
-    @hikari_test_helpers.timeout()
-    async def test__heartbeat(self, client):
-        client._last_heartbeat_sent = 5
+        assert shield.call_count == 2
+        shield.assert_has_calls([mock.call(heartbeat_task), mock.call(poll_events_task)])
+        first_completed.assert_called_once_with(
+            client._handshake_event.wait.return_value, shielded_heartbeat_task, shielded_poll_events_task
+        )
+
+    async def test__connect_when_reconnecting(self, client, http_settings, proxy_settings):
+        ws = mock.AsyncMock()
+        ws.receive_json.return_value = {"op": 10, "d": {"heartbeat_interval": 10}}
+        client._transport_compression = True
+        client._shard_id = 20
+        client._gateway_url = "wss://somewhere.com?somewhere=false"
+        client._resume_gateway_url = "wss://notsomewhere.com?somewhere=true"
+        client._token = "sometoken"
         client._logger = mock.Mock()
-        client._closing_event = mock.Mock(is_set=mock.Mock(return_value=False))
-        client._closed_event = mock.Mock(is_set=mock.Mock(return_value=False))
-        client._send_heartbeat = mock.AsyncMock()
+        client._handshake_event = mock.Mock()
+        client._seq = 1234
+        client._session_id = "some session id"
 
-        with mock.patch.object(time, "monotonic", return_value=10):
-            with mock.patch.object(asyncio, "wait_for", side_effect=[asyncio.TimeoutError, None]) as wait_for:
-                assert await client._heartbeat(20) is False
+        heartbeat_task = object()
+        poll_events_task = object()
+        shielded_heartbeat_task = object()
+        shielded_poll_events_task = object()
 
-        wait_for.assert_awaited_with(client._closing_event.wait(), timeout=20)
+        stack = contextlib.ExitStack()
+        create_task = stack.enter_context(
+            mock.patch.object(asyncio, "create_task", side_effect=[heartbeat_task, poll_events_task])
+        )
+        shield = stack.enter_context(
+            mock.patch.object(asyncio, "shield", side_effect=[shielded_heartbeat_task, shielded_poll_events_task])
+        )
+        first_completed = stack.enter_context(mock.patch.object(aio, "first_completed"))
+        log_filterer = stack.enter_context(mock.patch.object(shard, "_log_filterer"))
+        send_json = stack.enter_context(mock.patch.object(shard.GatewayShardImpl, "_send_json"))
+        heartbeat = stack.enter_context(mock.patch.object(shard.GatewayShardImpl, "_heartbeat", new=mock.Mock()))
+        poll_events = stack.enter_context(mock.patch.object(shard.GatewayShardImpl, "_poll_events", new=mock.Mock()))
+        gateway_transport_connect = stack.enter_context(
+            mock.patch.object(shard._GatewayTransport, "connect", return_value=ws)
+        )
+        stack.enter_context(mock.patch.object(shard, "_VERSION", new=400))
 
-    @hikari_test_helpers.timeout()
-    async def test__heartbeat_when_zombie(self, client):
-        client._last_heartbeat_sent = 10
-        client._logger = mock.Mock()
+        with stack:
+            assert await client._connect() == (heartbeat_task, poll_events_task)
+
+        log_filterer.assert_called_once_with("sometoken")
+        gateway_transport_connect.assert_called_once_with(
+            http_settings=http_settings,
+            log_filterer=log_filterer.return_value,
+            logger=client._logger,
+            proxy_settings=proxy_settings,
+            transport_compression=True,
+            url="wss://notsomewhere.com?somewhere=true&v=400&encoding=json&compress=zlib-stream",
+        )
+        client._event_factory.deserialize_connected_event.assert_called_once_with(client)
+        client._event_manager.dispatch.assert_called_once_with(
+            client._event_factory.deserialize_connected_event.return_value
+        )
 
-        with mock.patch.object(time, "monotonic", return_value=5):
-            with mock.patch.object(asyncio, "wait_for") as wait_for:
-                assert await client._heartbeat(20) is True
+        assert create_task.call_count == 2
+        create_task.assert_has_calls(
+            [
+                mock.call(heartbeat.return_value, name="heartbeat (shard 20)"),
+                mock.call(poll_events.return_value, name="poll events (shard 20)"),
+            ],
+        )
+        heartbeat.assert_called_once_with(0.01)
 
-        wait_for.assert_not_called()
+        ws.receive_json.assert_awaited_once_with()
+        send_json.assert_called_once_with(
+            {
+                "op": 6,
+                "d": {
+                    "token": "sometoken",
+                    "seq": 1234,
+                    "session_id": "some session id",
+                },
+            }
+        )
 
-    async def test__resume(self, client):
-        client._token = "token"
-        client._seq = 123
-        client._session_id = 456
-        client._send_json = mock.AsyncMock()
+        assert shield.call_count == 2
+        shield.assert_has_calls([mock.call(heartbeat_task), mock.call(poll_events_task)])
+        first_completed.assert_called_once_with(
+            client._handshake_event.wait.return_value, shielded_heartbeat_task, shielded_poll_events_task
+        )
 
-        await client._resume()
+    async def test__connect_when_op_received_is_not_HELLO(self, client):
+        ws = mock.AsyncMock()
+        ws.receive_json.return_value = {"op": 0, "d": {"not": "hello"}}
+        client._gateway_url = "somewhere.com"
+        client._logger = mock.Mock()
+        client._handshake_event = object()
 
-        expected_json = {
-            "op": 6,
-            "d": {"token": "token", "seq": 123, "session_id": 456},
-        }
-        client._send_json.assert_awaited_once_with(expected_json)
+        stack = contextlib.ExitStack()
+        stack.enter_context(pytest.raises(errors.GatewayError))
+        gateway_transport_connect = stack.enter_context(
+            mock.patch.object(shard._GatewayTransport, "connect", return_value=ws)
+        )
 
-    @pytest.mark.skip("TODO")
-    async def test__run(self, client):
-        ...
+        with stack:
+            assert await client._connect()
+
+        gateway_transport_connect.return_value.send_close.assert_awaited_once_with(
+            code=1002, message=b"Expected HELLO op"
+        )
 
     @pytest.mark.skip("TODO")
-    async def test__run_once(self, client):
+    async def test__keep_alive(self, client):
         ...
 
     async def test__send_heartbeat(self, client):
-        client._send_json = mock.AsyncMock()
         client._last_heartbeat_sent = 0
         client._seq = 10
 
-        with mock.patch.object(time, "monotonic", return_value=200):
-            await client._send_heartbeat()
+        with mock.patch.object(shard.GatewayShardImpl, "_send_json") as send_json:
+            with mock.patch.object(time, "monotonic", return_value=200):
+                await client._send_heartbeat()
 
-        client._send_json.assert_awaited_once_with({"op": 1, "d": 10})
+        send_json.assert_awaited_once_with({"op": 1, "d": 10}, priority=True)
         assert client._last_heartbeat_sent == 200
+
+    async def test__poll_events_on_dispatch(self, client):
+        payload = {"op": 0, "t": "SOMETHING", "d": {"some": "test"}, "s": 101}
+
+        client._ws = mock.Mock(receive_json=mock.AsyncMock(side_effect=[payload, RuntimeError]))
+        client._seq = 1000
+        client._event_manager.consume_raw_event = mock.Mock(side_effect=[LookupError])
+        client._handshake_event = mock.Mock()
+
+        with pytest.raises(RuntimeError):
+            await client._poll_events()
+
+        assert client._ws.receive_json.await_count == 2
+        assert client._seq == 101
+        client._event_manager.consume_raw_event.assert_called_once_with("SOMETHING", client, {"some": "test"})
+        client._handshake_event.set.assert_not_called()
+
+    async def test__poll_events_on_dispatch_when_READY(self, client):
+        data = {
+            "v": 10,
+            "session_id": 100001,
+            "resume_gateway_url": "testing_endpoint",
+            "user": {
+                "id": 123,
+                "username": "davfsa",
+                "discriminator": "7026",
+            },
+            "guilds": ["1"] * 100,
+        }
+
+        payload = {"op": 0, "t": "READY", "d": data, "s": 101}
+
+        client._ws = mock.Mock(receive_json=mock.AsyncMock(side_effect=[payload, RuntimeError]))
+        client._seq = None
+        client._session_id = None
+        client._resume_gateway_url = None
+        client._user_id = None
+        client._event_manager.consume_raw_event = mock.Mock(side_effect=[LookupError])
+        client._handshake_event = mock.Mock()
+
+        with pytest.raises(RuntimeError):
+            await client._poll_events()
+
+        assert client._ws.receive_json.await_count == 2
+        assert client._seq == 101
+        assert client._resume_gateway_url == "testing_endpoint"
+        assert client._session_id == 100001
+        assert client._user_id == 123
+        client._event_manager.consume_raw_event.assert_called_once_with("READY", client, data)
+        client._handshake_event.set.assert_called_once_with()
+
+    async def test__poll_events_on_dispatch_when_RESUMED(self, client):
+        payload = {"op": 0, "t": "RESUMED", "d": {"some": "test"}, "s": 101}
+
+        client._ws = mock.Mock(receive_json=mock.AsyncMock(side_effect=[payload, RuntimeError]))
+        client._seq = 1000
+        client._event_manager.consume_raw_event = mock.Mock(side_effect=[LookupError])
+        client._handshake_event = mock.Mock()
+
+        with pytest.raises(RuntimeError):
+            await client._poll_events()
+
+        assert client._ws.receive_json.await_count == 2
+        assert client._seq == 101
+        client._event_manager.consume_raw_event.assert_called_once_with("RESUMED", client, {"some": "test"})
+        client._handshake_event.set.assert_called_once_with()
+
+    async def test__poll_events_on_heartbeat_ack(self, client):
+        payload = {"op": 11}
+
+        client._ws = mock.Mock(receive_json=mock.AsyncMock(side_effect=[payload, RuntimeError]))
+        client._heartbeat_latency = 0
+        client._last_heartbeat_ack_received = 0
+        client._last_heartbeat_sent = 1.5
+        client._handshake_event = mock.Mock()
+
+        with mock.patch.object(time, "monotonic", return_value=3):
+            with pytest.raises(RuntimeError):
+                await client._poll_events()
+
+        assert client._ws.receive_json.await_count == 2
+        assert client._last_heartbeat_ack_received == 3
+        assert client._heartbeat_latency == 1.5
+        client._handshake_event.set.assert_not_called()
+
+    async def test__poll_events_on_heartbeat(self, client):
+        payload = {"op": 1}
+
+        client._ws = mock.Mock(receive_json=mock.AsyncMock(side_effect=[payload, RuntimeError]))
+        client._handshake_event = mock.Mock()
+
+        with mock.patch.object(shard.GatewayShardImpl, "_send_heartbeat") as send_heartbeat:
+            with pytest.raises(RuntimeError):
+                await client._poll_events()
+
+        assert client._ws.receive_json.await_count == 2
+        send_heartbeat.assert_awaited_once_with()
+        client._handshake_event.set.assert_not_called()
+
+    async def test__poll_events_on_reconnect(self, client):
+        payload = {"op": 7}
+
+        client._ws = mock.Mock(receive_json=mock.AsyncMock(side_effect=[payload, RuntimeError]))
+        client._handshake_event = mock.Mock()
+
+        await client._poll_events()
+
+        assert client._ws.receive_json.await_count == 1
+        client._handshake_event.set.assert_not_called()
+
+    async def test__poll_events_on_invalid_session_when_can_resume(self, client):
+        payload = {"op": 9, "d": True}
+
+        client._seq = 123
+        client._session_id = 456
+        client._ws = mock.Mock(receive_json=mock.AsyncMock(side_effect=[payload, RuntimeError]))
+        client._handshake_event = mock.Mock()
+
+        await client._poll_events()
+
+        assert client._ws.receive_json.await_count == 1
+        assert client._seq == 123
+        assert client._session_id == 456
+        client._handshake_event.set.assert_not_called()
+
+    async def test__poll_events_on_invalid_session_when_cant_resume(self, client):
+        payload = {"op": 9, "d": False}
+
+        client._seq = 123
+        client._session_id = 456
+        client._ws = mock.Mock(receive_json=mock.AsyncMock(side_effect=[payload, RuntimeError]))
+        client._handshake_event = mock.Mock()
+
+        await client._poll_events()
+
+        assert client._ws.receive_json.await_count == 1
+        assert client._seq is None
+        assert client._session_id is None
+        client._handshake_event.set.assert_not_called()
+
+    async def test__poll_events_on_unknown_op(self, client):
+        payload = {"op": 69, "d": "DATA"}
+
+        client._logger = mock.Mock()
+        client._ws = mock.Mock(receive_json=mock.AsyncMock(side_effect=[payload, RuntimeError]))
+        client._handshake_event = mock.Mock()
+
+        with pytest.raises(RuntimeError):
+            await client._poll_events()
+
+        assert client._ws.receive_json.await_count == 2
+        client._logger.log.assert_called_once_with(ux.TRACE, "unknown opcode %s received, it will be ignored...", 69)
+        client._handshake_event.set.assert_not_called()
diff --git a/tests/hikari/internal/test_signals.py b/tests/hikari/internal/test_signals.py
new file mode 100644
index 00000000..2688d4fa
--- /dev/null
+++ b/tests/hikari/internal/test_signals.py
@@ -0,0 +1,93 @@
+# -*- coding: utf-8 -*-
+# Copyright (c) 2020 Nekokatt
+# Copyright (c) 2021-present davfsa
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy
+# of this software and associated documentation files (the "Software"), to deal
+# in the Software without restriction, including without limitation the rights
+# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+# copies of the Software, and to permit persons to whom the Software is
+# furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all
+# copies or substantial portions of the Software.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+import contextlib
+import signal
+
+import mock
+import pytest
+
+from hikari import errors
+from hikari.internal import signals
+
+
+def test__raise_interrupt():
+    with pytest.raises(errors.HikariInterrupt):
+        signals._raise_interrupt(1)
+
+
+@pytest.mark.parametrize("trace", [True, False])
+def test__interrupt_handler(trace):
+    loop = mock.Mock()
+
+    with mock.patch.object(signals, "_LOGGER", new=mock.Mock(isEnabledFor=mock.Mock(return_value=trace))):
+        handler = signals._interrupt_handler(loop)
+
+        handler(1, None)
+
+    loop.call_soon_threadsafe.assert_called_once_with(signals._raise_interrupt, 1)
+
+
+class TestHandleInterrupt:
+    def test_behaviour(self):
+        loop = object()
+
+        stack = contextlib.ExitStack()
+        register_signal_handler = stack.enter_context(mock.patch.object(signal, "signal"))
+        interrupt_handler = stack.enter_context(mock.patch.object(signals, "_interrupt_handler"))
+        stack.enter_context(mock.patch.object(signal, "SIGINT", new=2, create=True))
+        stack.enter_context(mock.patch.object(signal, "SIGTERM", new=15, create=True))
+        stack.enter_context(mock.patch.object(signals, "_INTERRUPT_SIGNALS", ("SIGINT", "SIGTERM", "UNIMPLEMENTED")))
+
+        with stack:
+            with signals.handle_interrupts(True, loop, True):
+                interrupt_handler.assert_called_once_with(loop)
+
+                assert register_signal_handler.call_count == 2
+                register_signal_handler.assert_has_calls(
+                    [
+                        mock.call(2, interrupt_handler.return_value),
+                        mock.call(15, interrupt_handler.return_value),
+                    ]
+                )
+
+                register_signal_handler.reset_mock()
+
+        assert register_signal_handler.call_count == 2
+        register_signal_handler.assert_has_calls([mock.call(2, signal.SIG_DFL), mock.call(15, signal.SIG_DFL)])
+
+    def test_when_disabled(self):
+        with mock.patch.object(signal, "signal") as register_signal_handler:
+            with signals.handle_interrupts(False, object(), True):
+                register_signal_handler.assert_not_called()
+
+        register_signal_handler.assert_not_called()
+
+    def test_when_propagate_interrupt(self):
+        with mock.patch.object(signal, "signal"):
+            with pytest.raises(errors.HikariInterrupt):  # noqa:  PT012 - raises block should contain a single statement
+                with signals.handle_interrupts(True, object(), True):
+                    raise errors.HikariInterrupt(1, "t")
+
+    def test_when_not_propagate_interrupt(self):
+        with mock.patch.object(signal, "signal"):
+            with signals.handle_interrupts(True, object(), False):
+                raise errors.HikariInterrupt(1, "t")