fix(zip): handle when unsubscribe is called from within next

packages/rxjs/spec/observables/zip-spec.ts

@@ -1,6 +1,6 @@
 /** @prettier */
 import { expect } from 'chai';
-import { queueScheduler as rxQueueScheduler, zip, from, scheduled } from 'rxjs';
+import { queueScheduler as rxQueueScheduler, zip, from, scheduled, of, config, concat, NEVER, first, delay, filter, GlobalConfig } from 'rxjs';
 import { TestScheduler } from 'rxjs/testing';
 import { observableMatcher } from '../helpers/observableMatcher';
 
@@ -630,4 +630,28 @@ describe('zip', () => {
 
     expect(results).to.deep.equal(['done']);
   });
+
+  describe('with a registered notificaiton handler', () => {
+    let onStoppedNotification: GlobalConfig['onStoppedNotification'];
+
+    beforeEach(() => onStoppedNotification = config.onStoppedNotification);
+
+    afterEach(() => config.onStoppedNotification = onStoppedNotification);
+
+    it('should handle when unsubscribing from within a next handler', (done) => {
+      let error: any = null;
+
+      config.onStoppedNotification = (notification) => {
+        if (notification.kind === 'E') {
+          error = notification.error;
+
+          done(notification.error);
+        }
+      };
+
+      const source$ = concat(of(1), NEVER);
+
+      zip(source$).pipe(first(), delay(1), filter(() => error === null)).subscribe(() => done());
+    });
+  });
 });

packages/rxjs/src/internal/observable/zip.ts

@@ -88,7 +88,7 @@ export function zip(...args: unknown[]): Observable<unknown> {
                   // If any one of the sources is both complete and has an empty buffer
                   // then we complete the result. This is because we cannot possibly have
                   // any more values to zip together.
-                  if (buffers.some((buffer, i) => !buffer.length && completed[i])) {
+                  if (buffers?.some((buffer, i) => !buffer.length && completed[i])) {
                     destination.complete();
                   }
                 }