Merge pull request #1177 from browserstack/fix/sdk-7399-o11y-flush-skips-tests

fix(o11y): batch the event flush so command-heavy specs stop hitting spec_timeout [SDK-7399]

Author: hamza-browserstack

bin/testObservability/cypress/index.js

@@ -50,7 +50,8 @@ const shouldSkipCommand = (command) => {
   if (!Cypress.env('BROWSERSTACK_O11Y_LOGS')) {
     return true;
   }
-  return command.attributes.name == 'log' || (command.attributes.name == 'task' && (['test_observability_platform_details', 'test_observability_step', 'test_observability_command', 'browserstack_log', 'test_observability_log'].some(event => command.attributes.args.includes(event))));
+  /* the batch task is filtered too, else each dispatch refills the queue */
+  return command.attributes.name == 'log' || (command.attributes.name == 'task' && (['test_observability_platform_details', 'test_observability_step', 'test_observability_command', 'test_observability_batch', 'browserstack_log', 'test_observability_log'].some(event => command.attributes.args.includes(event))));
 }
 
 Cypress.on('log:changed', (attrs) => {
@@ -339,20 +340,91 @@ Cypress.Commands.add('fatal', (message, file) => {
   });
 });
 
+/* console.warn, not cy.task — a diagnostic must not use the mechanism it reports on */
+const warnFlushFailure = (stage, err) => {
+  try {
+    console.warn(`BrowserStack Test Observability: suppressed ${stage} error, event(s) dropped: ${err && err.message ? err.message : err}`);
+  } catch (e) { /* logging must never throw either */ }
+};
+
+/*
+ * [SDK-7399] One cy.task per drain, not per event. Each round-trip costs ~0.8s on a
+ * remote terminal, so a command-heavy spec spent minutes in afterEach, exceeded
+ * spec_timeout and was killed — unrun tests then reported as skipped, with nothing
+ * thrown. 600 events: 581s as 600 calls, 109s as one.
+ * Stays on cy.task: cy.now('task') throws on Cypress 14, so it would "fix" this by
+ * delivering nothing. The try/catch is only a backstop for synchronous throws — a
+ * cy.task failure surfaces later, while the command queue drains.
+ */
+
+/* Remote terminal: a single cy.task payload of 768KB succeeds, 1MB fails.
+ * Split well under that; drop only what cannot be sent at all. The two must stay
+ * distinct — an event between them still sends on its own. */
+const MAX_BATCH_CHARS = 512 * 1024;
+const MAX_EVENT_CHARS = 768 * 1024;
+
+const flushEventsQueue = () => {
+  try {
+    const queued = eventsQueue;
+    eventsQueue = []; /* cleared before dispatch so a throw cannot replay these events */
+    if (queued.length === 0) return;
+
+    let batch = [];
+    let batchChars = 0;
+
+    const sendBatch = () => {
+      if (batch.length === 0) return;
+      const toSend = batch;
+      batch = [];
+      batchChars = 0;
+      try {
+        /* every push site uses { log: false }, so per-event options are not forwarded */
+        cy.task('test_observability_batch', toSend, { log: false });
+      } catch (e) {
+        warnFlushFailure(`batch dispatch of ${toSend.length} event(s)`, e);
+      }
+    };
+
+    queued.forEach(event => {
+      try {
+        const payload = sanitizeForTask(event.data);
+        if (payload === null) {
+          warnFlushFailure(`unserializable payload for '${event.task}'`,
+            new Error('event skipped'));
+          return;
+        }
+        const size = JSON.stringify(payload).length;
+        if (size > MAX_EVENT_CHARS) {
+          /* past the largest size measured to send; 768KB-1MB is untested, so skip */
+          warnFlushFailure(`event too large to send for '${event.task}' (${size} chars)`,
+            new Error('event skipped'));
+          return;
+        }
+        /* oversized-but-sendable: let it travel alone */
+        if (batch.length > 0 && batchChars + size > MAX_BATCH_CHARS) sendBatch();
+        batch.push({ task: event.task, data: payload });
+        batchChars += size;
+        if (batchChars >= MAX_BATCH_CHARS) sendBatch();
+      } catch (e) {
+        warnFlushFailure(`preparing '${event.task}'`, e); /* skip one event, not the rest */
+      }
+    });
+
+    sendBatch();
+  } catch (e) {
+    warnFlushFailure('queue flush', e);
+    eventsQueue = [];
+  }
+};
+
 beforeEach(() => {
   /* browserstack internal helper hook */
 
   if (!Cypress.env('BROWSERSTACK_O11Y_LOGS')) {
     return;
   }
 
-  if (eventsQueue.length > 0) {
-    eventsQueue.forEach(event => {
-      const payload = sanitizeForTask(event.data);
-      if (payload !== null) cy.task(event.task, payload, event.options);
-    });
-  }
-  eventsQueue = [];
+  flushEventsQueue();
   testRunStarted = true;
 });
 
@@ -362,13 +434,6 @@ afterEach(function() {
     return;
   }
 
-  if (eventsQueue.length > 0) {
-    eventsQueue.forEach(event => {
-      const payload = sanitizeForTask(event.data);
-      if (payload !== null) cy.task(event.task, payload, event.options);
-    });
-  }
-  
-  eventsQueue = [];
+  flushEventsQueue();
   testRunStarted = false;
 });

bin/testObservability/plugin/index.js

@@ -11,6 +11,13 @@ const browserstackTestObservabilityPlugin = (on, config, callbacks) => {
 
   connectIPCClient(config);
 
+  const IPC_EVENT_FOR_TASK = {
+    test_observability_log: IPC_EVENTS.LOG,
+    test_observability_command: IPC_EVENTS.COMMAND,
+    test_observability_platform_details: IPC_EVENTS.PLATFORM_DETAILS,
+    test_observability_step: IPC_EVENTS.CUCUMBER,
+  };
+
   on('task', {
     test_observability_log(log) {
       ipc.of.browserstackTestObservability.emit(IPC_EVENTS.LOG, log);
@@ -27,6 +34,21 @@ const browserstackTestObservabilityPlugin = (on, config, callbacks) => {
     test_observability_step(log) {
       ipc.of.browserstackTestObservability.emit(IPC_EVENTS.CUCUMBER, log);
       return null;
+    },
+    /* [SDK-7399] One task per flush instead of one per event — see cypress/index.js.
+     * Fans out to the same IPC events; the per-event tasks stay for back-compat. */
+    test_observability_batch(events) {
+      if (!Array.isArray(events)) return null;
+      events.forEach((event) => {
+        try {
+          const ipcEvent = event && IPC_EVENT_FOR_TASK[event.task];
+          if (!ipcEvent) return;
+          ipc.of.browserstackTestObservability.emit(ipcEvent, event.data);
+        } catch (e) {
+          /* one bad entry must not drop the rest */
+        }
+      });
+      return null;
     }
   });
 

test/unit/bin/testObservability/batchFlush.js

@@ -0,0 +1,85 @@
+'use strict';
+const chai = require('chai');
+const expect = chai.expect;
+const sinon = require('sinon');
+const proxyquire = require('proxyquire');
+
+// Regression guard for SDK-7399. The flush used to issue one cy.task per queued event,
+// and each round-trip costs ~0.8s on a remote terminal, so a command-heavy spec spent
+// minutes in afterEach, exceeded spec_timeout and was killed — every test that had not
+// run yet was then reported as skipped. Nothing threw, so a passing spec alone cannot
+// distinguish "delivered" from "silently dropped"; these tests pin the fan-out instead.
+describe('SDK-7399 batched observability flush', () => {
+  let emit, ipcStub, tasks, plugin;
+
+  beforeEach(() => {
+    emit = sinon.stub();
+    ipcStub = { of: { browserstackTestObservability: { emit } } };
+    plugin = proxyquire('../../../../bin/testObservability/plugin', {
+      'node-ipc': ipcStub,
+      './ipcClient': { connectIPCClient: () => {} },
+    });
+    tasks = null;
+    const on = (name, handlers) => { if (name === 'task') tasks = handlers; };
+    plugin(on, { env: {} });
+  });
+
+  afterEach(() => sinon.restore());
+
+  it('registers the batch task alongside the per-event tasks', () => {
+    expect(tasks).to.have.property('test_observability_batch');
+    // per-event tasks must stay registered: an older browser bundle may still call them
+    expect(tasks).to.have.property('test_observability_log');
+    expect(tasks).to.have.property('test_observability_command');
+    expect(tasks).to.have.property('test_observability_platform_details');
+    expect(tasks).to.have.property('test_observability_step');
+  });
+
+  it('fans every queued task type out to its own IPC event, in order', () => {
+    tasks.test_observability_batch([
+      { task: 'test_observability_log', data: { m: 1 } },
+      { task: 'test_observability_command', data: { m: 2 } },
+      { task: 'test_observability_platform_details', data: { m: 3 } },
+      { task: 'test_observability_step', data: { m: 4 } },
+    ]);
+
+    expect(emit.callCount).to.equal(4);
+    expect(emit.getCalls().map(c => c.args[1].m)).to.deep.equal([1, 2, 3, 4]);
+    // four distinct IPC events, i.e. no type collapsed onto another
+    expect(new Set(emit.getCalls().map(c => c.args[0])).size).to.equal(4);
+  });
+
+  it('emits one IPC event per entry for a large batch', () => {
+    const batch = [];
+    for (let i = 0; i < 600; i++) {
+      batch.push({ task: 'test_observability_log', data: { i } });
+    }
+    tasks.test_observability_batch(batch);
+    expect(emit.callCount).to.equal(600);
+  });
+
+  it('skips an unknown task name but still delivers the rest of the batch', () => {
+    tasks.test_observability_batch([
+      { task: 'not_a_real_task', data: { m: 'x' } },
+      { task: 'test_observability_log', data: { m: 'kept' } },
+    ]);
+    expect(emit.callCount).to.equal(1);
+    expect(emit.firstCall.args[1].m).to.equal('kept');
+  });
+
+  it('never throws on malformed input', () => {
+    expect(() => tasks.test_observability_batch(undefined)).to.not.throw();
+    expect(() => tasks.test_observability_batch({})).to.not.throw();
+    expect(() => tasks.test_observability_batch([null, undefined, 1, 'x'])).to.not.throw();
+    expect(emit.callCount).to.equal(0);
+  });
+
+  it('one failing emit does not drop the remaining entries', () => {
+    emit.onFirstCall().throws(new Error('ipc down'));
+    tasks.test_observability_batch([
+      { task: 'test_observability_log', data: { m: 1 } },
+      { task: 'test_observability_log', data: { m: 2 } },
+    ]);
+    expect(emit.callCount).to.equal(2);
+  });
+});

test/unit/bin/testObservability/batchThresholds.js

@@ -0,0 +1,121 @@
+'use strict';
+const chai = require('chai');
+const expect = chai.expect;
+const sinon = require('sinon');
+
+// Guards the threshold arithmetic in bin/testObservability/cypress/index.js.
+// SDK-7399 review finding: the batch-split figure (512KB) had been reused as the
+// single-event drop ceiling, so events in the 512-768KB band — which a single cy.task
+// was measured to carry — were silently discarded. These cases pin the split, the
+// oversized-but-sendable path, and the drop boundary.
+describe('SDK-7399 flush thresholds', () => {
+  const KB = 1024;
+  let taskSpy, afterEachCb, commandStartCb;
+
+  // The browser-side file registers listeners and hooks at require time, so the Cypress
+  // globals have to exist first. Capture the pieces the flush needs.
+  const loadBrowserSide = () => {
+    const listeners = {};
+    taskSpy = sinon.stub().returns(undefined);
+
+    global.cy = { task: taskSpy, now: sinon.stub() };
+    global.Cypress = {
+      on: (evt, cb) => { listeners[evt] = cb; },
+      env: (k) => (k === 'BROWSERSTACK_O11Y_LOGS' ? 'true' : undefined),
+      Commands: { add: () => {}, overwrite: () => {} },
+      browser: { name: 'chrome', majorVersion: '136' },
+      platform: 'win32',
+      version: '14.3.3',
+      mocha: { getRunner: () => ({ suite: { ctx: { currentTest: { title: 't' } } } }) },
+    };
+    global.beforeEach = () => {};
+    global.afterEach = (cb) => { afterEachCb = cb; };
+
+    delete require.cache[require.resolve('../../../../bin/testObservability/cypress')];
+    require('../../../../bin/testObservability/cypress');
+    commandStartCb = listeners['command:start'];
+  };
+
+  // One queued event whose serialized payload is ~sizeKB, via a command arg.
+  const queueEventOfSize = (sizeKB) => {
+    commandStartCb({ attributes: { id: 'c1', name: 'type', args: ['x'.repeat(sizeKB * KB)] } });
+  };
+
+  beforeEach(loadBrowserSide);
+
+  afterEach(() => {
+    sinon.restore();
+    delete global.cy; delete global.Cypress;
+    delete global.beforeEach; delete global.afterEach;
+  });
+
+  const batchesSent = () =>
+    taskSpy.getCalls()
+      .filter(c => c.args[0] === 'test_observability_batch')
+      .map(c => c.args[1]);
+
+  it('sends small events together in a single batch', () => {
+    queueEventOfSize(1);
+    queueEventOfSize(1);
+    afterEachCb();
+
+    const batches = batchesSent();
+    expect(batches.length).to.equal(1);
+    expect(batches[0].length).to.be.greaterThan(1);
+  });
+
+  it('dispatches a 600KB event alone rather than dropping it (the regression)', () => {
+    // command:start queues two events: the command itself, plus small platform details —
+    // so the big one is expected in a batch of its own, with the small one following.
+    queueEventOfSize(600);
+    afterEachCb();
+
+    const batches = batchesSent();
+    const carrying = batches.filter(b =>
+      b.some(e => JSON.stringify(e.data).length > 512 * KB));
+    expect(carrying.length, 'the 512-768KB band must still be delivered').to.equal(1);
+    expect(carrying[0].length, 'oversized-but-sendable event travels alone').to.equal(1);
+  });
+
+  it('keeps a 600KB event out of the batch holding the small ones', () => {
+    queueEventOfSize(1);
+    queueEventOfSize(600);
+    afterEachCb();
+
+    const batches = batchesSent();
+    expect(batches.length).to.be.greaterThan(1);
+    batches.forEach(b => {
+      const chars = b.reduce((n, e) => n + JSON.stringify(e.data).length, 0);
+      // a multi-event batch stays under the split figure; a lone event may exceed it
+      if (b.length > 1) expect(chars).to.be.at.most(512 * KB);
+    });
+  });
+
+  it('skips an event past the largest measured-safe size', () => {
+    queueEventOfSize(900);
+    afterEachCb();
+
+    batchesSent().forEach(b => {
+      b.forEach(e => {
+        expect(JSON.stringify(e.data).length).to.be.at.most(768 * KB);
+      });
+    });
+  });
+
+  it('never assembles a multi-event batch beyond the split figure', () => {
+    for (let i = 0; i < 12; i++) queueEventOfSize(64);
+    afterEachCb();
+
+    batchesSent().forEach(b => {
+      if (b.length > 1) {
+        const chars = b.reduce((n, e) => n + JSON.stringify(e.data).length, 0);
+        expect(chars).to.be.at.most(512 * KB);
+      }
+    });
+  });
+
+  it('sends nothing when the queue is empty', () => {
+    afterEachCb();
+    expect(batchesSent().length).to.equal(0);
+  });
+});