Migrate waitForConnectionToBeUsable() to react to pushed connection event state changes

.versions/ide-sidecar.txt

@@ -1 +1 @@
-v0.134.1
+v0.139.0

CHANGELOG.md

@@ -4,6 +4,9 @@ All notable changes to this extension will be documented in this file.
 
 ## Unreleased
 
+### Changed
+- React to websocket events pushed from sidecar instead of polling for some connection change monitoring.
+
 ## 0.23.3
 
 ### Fixed

src/authn/ccloudProvider.test.ts

@@ -59,7 +59,7 @@ describe("ConfluentCloudAuthProvider", () => {
 
     // assume the connection is immediately usable for most tests
     sandbox
-      .stub(connections, "waitForConnectionToBeUsable")
+      .stub(connections, "waitForConnectionToBeStable")
       .resolves(TEST_AUTHENTICATED_CCLOUD_CONNECTION);
 
     // don't handle the progress notification, openExternal, etc in this test suite

src/authn/ccloudProvider.ts

@@ -13,7 +13,7 @@ import {
   createCCloudConnection,
   deleteCCloudConnection,
   getCCloudConnection,
-  waitForConnectionToBeUsable,
+  waitForConnectionToBeStable,
 } from "../sidecar/connections";
 import { getStorageManager } from "../storage";
 import { SecretStorageKeys } from "../storage/constants";
@@ -152,7 +152,7 @@ export class ConfluentCloudAuthProvider implements vscode.AuthenticationProvider
       throw e;
     }
 
-    const authenticatedConnection = await waitForConnectionToBeUsable(CCLOUD_CONNECTION_ID);
+    const authenticatedConnection = await waitForConnectionToBeStable(CCLOUD_CONNECTION_ID);
     if (!authenticatedConnection) {
       throw new Error("CCloud connection failed to become usable after authentication.");
     }

src/directConnectManager.test.ts

@@ -71,7 +71,7 @@ describe("DirectConnectionManager behavior", () => {
       .stub(connections, "tryToUpdateConnection")
       .resolves({} as any);
     // assume the connection is immediately usable for most tests
-    sandbox.stub(connections, "waitForConnectionToBeUsable").resolves(TEST_DIRECT_CONNECTION);
+    sandbox.stub(connections, "waitForConnectionToBeStable").resolves(TEST_DIRECT_CONNECTION);
   });
 
   afterEach(() => {

src/directConnectManager.ts

@@ -18,7 +18,7 @@ import {
   tryToCreateConnection,
   tryToDeleteConnection,
   tryToUpdateConnection,
-  waitForConnectionToBeUsable,
+  waitForConnectionToBeStable,
 } from "./sidecar/connections";
 import { SecretStorageKeys } from "./storage/constants";
 import { DirectResourceLoader } from "./storage/directResourceLoader";
@@ -218,6 +218,7 @@ export class DirectConnectionManager {
   ): Promise<{ connection: Connection | null; errorMessage: string | null }> {
     let connection: Connection | null = null;
     let errorMessage: string | null = null;
+
     try {
       connection = update
         ? await tryToUpdateConnection(spec)
@@ -230,7 +231,7 @@ export class DirectConnectionManager {
             title: `Waiting for "${connection.spec.name}" to be usable...`,
           },
           async () => {
-            await waitForConnectionToBeUsable(connectionId);
+            await waitForConnectionToBeStable(connectionId);
           },
         );
       }
@@ -298,7 +299,7 @@ export class DirectConnectionManager {
       // wait for all new connections to be created before checking their status
       await Promise.all(newConnectionPromises);
       // kick off background checks to ensure the new connections are usable
-      connectionIdsToCheck.map((id) => waitForConnectionToBeUsable(id));
+      connectionIdsToCheck.map((id) => waitForConnectionToBeStable(id));
       logger.debug(
         `created and checked ${connectionIdsToCheck.length} new connection(s), firing event`,
       );

src/emitters.ts

@@ -38,5 +38,4 @@ export const currentKafkaClusterChanged = new vscode.EventEmitter<KafkaCluster |
  */
 export const currentSchemaRegistryChanged = new vscode.EventEmitter<SchemaRegistry | null>();
 
-export const connectionLoading = new vscode.EventEmitter<ConnectionId>();
-export const connectionUsable = new vscode.EventEmitter<ConnectionId>();
+export const connectionStable = new vscode.EventEmitter<ConnectionId>();

src/extension.ts

@@ -74,7 +74,7 @@ import { createConfigChangeListener } from "./preferences/listener";
 import { updatePreferences } from "./preferences/updates";
 import { registerProjectGenerationCommand } from "./scaffold";
 import { getSidecarManager, sidecarOutputChannel } from "./sidecar";
-import { getCCloudAuthSession } from "./sidecar/connections";
+import { ConnectionStateWatcher, getCCloudAuthSession } from "./sidecar/connections";
 import { WebsocketManager } from "./sidecar/websocketManager";
 import { getStorageManager, StorageManager } from "./storage";
 import { SecretStorageKeys } from "./storage/constants";
@@ -224,6 +224,10 @@ async function _activateExtension(
   // reset the Docker credentials secret so `src/docker/configs.ts` can pull it fresh
   getStorageManager().deleteSecret(SecretStorageKeys.DOCKER_CREDS_SECRET_KEY);
 
+  // Watch for sidecar pushing connection state changes over websocket.
+  // (side effect of causing the watcher to be created)
+  ConnectionStateWatcher.getInstance();
+
   const directConnectionManager = DirectConnectionManager.getInstance();
   context.subscriptions.push(...directConnectionManager.disposables);
 

src/models/environment.test.ts

@@ -35,6 +35,9 @@ describe("EnvironmentTreeItem", () => {
       schemaRegistry: undefined,
     });
 
+    // Override isLoading to false due to no clusters
+    env.isLoading = false;
+
     const treeItem = new EnvironmentTreeItem(env);
 
     assert.deepStrictEqual(

src/models/environment.ts

@@ -45,7 +45,8 @@ export abstract class Environment implements IResourceBase {
   kafkaClusters!: KafkaCluster[];
   schemaRegistry?: SchemaRegistry | undefined;
 
-  // updated by the ResourceViewProvider from connectionLoading/connectionUsable events
+  // updated by the ResourceViewProvider from connectionUsable events
+  // (DirectEnvironment instances are constructed with isLoading = true)
   isLoading: boolean = false;
 
   get hasClusters(): boolean {
@@ -113,6 +114,12 @@ export class DirectEnvironment extends Environment {
     this.kafkaClusters = props.kafkaClusters;
     this.schemaRegistry = props.schemaRegistry;
     if (props.formConnectionType) this.formConnectionType = props.formConnectionType;
+
+    // newly born direct connections are loading unless we already have children.
+    // This will eventually mutate
+    // to false when the connection is stable and emitters.connectionStable fires through
+    // a real Rube Goldberg machine of events.
+    this.isLoading = !this.hasClusters;
   }
 
   get iconName(): IconNames {
@@ -174,6 +181,7 @@ export class EnvironmentTreeItem extends TreeItem {
     // user-facing properties
     this.description = isDirect(this.resource) ? "" : this.resource.id;
     this.iconPath = new ThemeIcon(this.resource.iconName);
+
     if (this.resource.isLoading) {
       this.iconPath = new ThemeIcon(IconNames.LOADING);
     } else if (isDirect(resource) && !resource.hasClusters) {

src/sidecar/connectionStatusUtils.test.ts

@@ -0,0 +1,73 @@
+// tests over isConnectionStable
+
+import * as assert from "assert";
+
+import {
+  TEST_CCLOUD_CONNECTION,
+  TEST_DIRECT_CONNECTION,
+  TEST_LOCAL_CONNECTION,
+} from "../../tests/unit/testResources/connection";
+
+import { ConnectedState, Status } from "../clients/sidecar/models";
+import { isConnectionStable } from "./connectionStatusUtils";
+
+describe("isConnectionStable", () => {
+  const testAuthStatus = { authentication: { status: Status.NoToken } };
+
+  it("ccloud connection tests", () => {
+    type CCloudConnectionStateAndResult = [ConnectedState, boolean];
+
+    // ccloud connection is stable if not in Attempting state
+    const testCases: CCloudConnectionStateAndResult[] = [
+      [ConnectedState.None, false],
+      [ConnectedState.Attempting, true],
+      [ConnectedState.Success, true],
+      [ConnectedState.Expired, true],
+      [ConnectedState.Failed, true],
+    ];
+
+    for (const [connectedState, expectedResult] of testCases) {
+      const testConnection = {
+        ...TEST_CCLOUD_CONNECTION,
+        status: {
+          ccloud: { state: connectedState },
+          ...testAuthStatus,
+        },
+      };
+      assert.strictEqual(isConnectionStable(testConnection), expectedResult);
+    }
+  });
+
+  it("direct connection tests", () => {
+    type LocalConnectionStatesAndResult = [ConnectedState, ConnectedState, boolean];
+
+    // direct connection is stable if neither kafka nor schema registry are attempting
+    const testCases: LocalConnectionStatesAndResult[] = [
+      [ConnectedState.None, ConnectedState.None, true],
+      [ConnectedState.None, ConnectedState.Success, true],
+      [ConnectedState.Success, ConnectedState.None, true],
+      [ConnectedState.Attempting, ConnectedState.None, false],
+      [ConnectedState.None, ConnectedState.Attempting, false],
+      [ConnectedState.Attempting, ConnectedState.Attempting, false],
+      [ConnectedState.Success, ConnectedState.Success, true],
+      [ConnectedState.Expired, ConnectedState.Expired, true],
+      [ConnectedState.Failed, ConnectedState.Failed, true],
+    ];
+
+    for (const [kafkaState, schemaRegistryState, expectedResult] of testCases) {
+      const testConnection = {
+        ...TEST_DIRECT_CONNECTION,
+        status: {
+          kafka_cluster: { state: kafkaState },
+          schema_registry: { state: schemaRegistryState },
+          ...testAuthStatus,
+        },
+      };
+      assert.strictEqual(isConnectionStable(testConnection), expectedResult);
+    }
+  });
+
+  it("should raise when asked about a local connection (not implemented yet)", () => {
+    assert.throws(() => isConnectionStable(TEST_LOCAL_CONNECTION));
+  });
+});

src/sidecar/connectionStatusUtils.ts

@@ -0,0 +1,58 @@
+// helpers for connection status testing, factored out for test spying
+
+import { Connection, ConnectionType } from "../clients/sidecar/models";
+import { Logger } from "../logging";
+
+const logger = new Logger("connectionStatusUtils");
+
+export function isConnectionStable(connection: Connection): boolean {
+  const type = connection.spec.type!;
+
+  switch (type) {
+    case ConnectionType.Ccloud:
+      return isCCloudConnectionStable(connection);
+    case ConnectionType.Direct:
+      return isDirectConnectionStable(connection);
+    default:
+      logger.warn(`isConnectionStable: unhandled connection type ${type}`);
+      throw new Error(`Unhandled connection type ${type}`);
+  }
+}
+
+function isCCloudConnectionStable(connection: Connection): boolean {
+  const ccloudStatus = connection.status.ccloud!;
+  const ccloudState = ccloudStatus.state;
+
+  const ccloudFailed = ccloudStatus.errors?.sign_in?.message;
+  if (ccloudFailed) {
+    logger.error(`isCCloudConnectionStable(): error: ${ccloudFailed}`);
+  }
+
+  const rv = ccloudState !== "NONE";
+  logger.debug(`isCCloudConnectionStable(): returning ${rv} based on state ${ccloudState}`);
+
+  return rv;
+}
+
+function isDirectConnectionStable(connection: Connection): boolean {
+  const status = connection.status;
+
+  for (const [entity, maybeError] of [
+    ["kafka", status.kafka_cluster?.errors?.sign_in?.message],
+    ["schema registry", status.schema_registry?.errors?.sign_in?.message],
+  ] as [string, string | undefined][]) {
+    if (maybeError) {
+      logger.error(`isDirectConnectionStable(): ${entity} error: ${maybeError}`);
+    }
+  }
+
+  const kafkaState = status.kafka_cluster?.state;
+  const schemaRegistryState = status.schema_registry?.state;
+
+  const rv = kafkaState !== "ATTEMPTING" && schemaRegistryState !== "ATTEMPTING";
+  logger.debug(
+    `isDirectConnectionStable(): returning ${rv} for connection ${connection.id} based on kafkaState ${kafkaState} and schemaRegistryState ${schemaRegistryState}`,
+  );
+
+  return rv;
+}