agents: fix infinite recursion on ~CommandStream

agents/grpc/src/command_stream.cc

@@ -31,6 +31,7 @@ CommandStream::~CommandStream() {
   nsuv::ns_mutex::scoped_lock lock(lock_);
   // try cancel and wait until OnDone is called
   if (!write_state_.done) {
+    cancelling_for_destruction_ = true;
     context_.TryCancel();
     do {
       uv_cond_wait(&on_done_cond_, lock_.base());
@@ -53,12 +54,11 @@ void CommandStream::OnDone(const Status& s) {
     nsuv::ns_mutex::scoped_lock lock(lock_);
     write_state_.done = true;
     uv_cond_signal(&on_done_cond_);
-    if (!obs) {
+    if (!obs || cancelling_for_destruction_) {
       return;
     }
   }
 
-  // Don't notify the observer if the stream was cancelled (destroyed)
   obs->on_command_stream_done(s);
 }
 

agents/grpc/src/command_stream.h

@@ -60,6 +60,7 @@ class CommandStream:
   TSQueue<grpcagent::CommandResponse> response_q_;
   nsuv::ns_mutex lock_;
   uv_cond_t on_done_cond_;
+  bool cancelling_for_destruction_ = false;
 };
 
 }  // namespace grpc

test/agents/test-grpc-basic.mjs

@@ -1,3 +1,3 @@
 // Flags: --expose-internals
 import { mustCall, mustSucceed } from '../common/index.mjs';
 import fixtures from '../common/fixtures.js';
@@ -126,6 +126,61 @@
   },
 });
 
+tests.push({
+  name: 'should reconnect after initial invalid NSOLID_GRPC',
+  test: async (getEnv, isSecure) => {
+    return new Promise((resolve) => {
+      const grpcServer = new GRPCServer({ tls: isSecure });
+      grpcServer.start(mustSucceed(async (port) => {
+        grpcServer.on('exit', mustCall((data) => {
+          checkExitData(data.msg, data.metadata, agentId, { code: 0, error: null, profile: '' });
+          grpcServer.close();
+          resolve();
+        }));
+
+        const correctEnv = getEnv(port, isSecure);
+        const env = {
+          NODE_DEBUG_NATIVE: 'nsolid_grpc_agent',
+          NSOLID_GRPC: '127.0.0.1:1',
+        };
+        if (correctEnv.NSOLID_GRPC_INSECURE) {
+          env.NSOLID_GRPC_INSECURE = correctEnv.NSOLID_GRPC_INSECURE;
+        }
+        if (correctEnv.NSOLID_GRPC_CERTS) {
+          env.NSOLID_GRPC_CERTS = correctEnv.NSOLID_GRPC_CERTS;
+        }
+
+        const opts = {
+          stdio: ['inherit', 'inherit', 'inherit', 'ipc'],
+          env,
+        };
+        const child = new TestClient([], opts);
+        const agentId = await child.id();
+
+        const config = {};
+        if (correctEnv.NSOLID_GRPC) {
+          config.grpc = correctEnv.NSOLID_GRPC;
+        }
+        if (correctEnv.NSOLID_SAAS) {
+          config.saas = correctEnv.NSOLID_SAAS;
+        }
+
+        grpcServer.on('command', async ({ agentId }) => {
+          // Verify the CommandStream is working by sending a command from server to client
+          const infoResult = await grpcServer.info(agentId);
+          assert.ok(infoResult);
+          const exit = await child.shutdown(0);
+          assert.ok(exit);
+          assert.strictEqual(exit.code, 0);
+          assert.strictEqual(exit.signal, null);
+        });
+
+        await child.config(config);
+      }));
+    });
+  },
+});
+
 const testConfigs = [
   {
     getEnv: (port, isSecure) => {