fix leaking websocket connections, subscription objects
A couple things going on here.
- We were registering operations on an inner observer, while orphaning the SubscriptionObserver we were subscribing to each observable in the subscription server. This diff makes sure we register the correct operation so that when the connection closes, we GC the SubscriptionObserver (which has the request context on it, including the GraphQL schema)
- We were not disposing of the observer correctly, because we were wrapping the observer in these "Subscription" objects for the event log and compute log case. In these cases, we needed to supply a callback to dispose of the observer when the connection gets closed and return it in the subscription call (or the __call__ method, in these cases). We can modify the __call__ method to return a disposable object and implement the dispose method to clean up the observer.
- For the compute logs and the event logs, we were registering callbacks to the instance by watching something. We had to unregister these callbacks as well, to remove the dangling reference to the subscription object. These unregister calls now happen in the newly created dispose method (see 2.). Had to create a new unsubscribe method on the ComputeLogManager abstract class and an end_watch_event_log method on DagsterInstance.
Test Plan: Created mem endpoint to monitor the objects in memory for SubscriptionObserver, ComputeLogSubscription, and PipelineRunObservableSubscribe. Saw that opening / closing tabs for various pages (run view) correctly returned those values to 0.
Reviewers: alangenfeld, max, sashank
Reviewed By: alangenfeld
Differential Revision: https://dagster.phacility.com/D8535