[@azure/cosmos] fix changefeed iterator TimeOutError bug

sdk/cosmosdb/cosmos/package.json

@@ -72,7 +72,7 @@
     "test:signoff": "npm run integration-test:node -- --fgrep \"nosignoff\" --invert",
     "unit-test": "npm run unit-test:node && npm run unit-test:browser",
     "unit-test:browser": "echo skipped",
-    "unit-test:node": "dev-tool run test:node-ts-input -- --timeout 100000 './test/internal/unit/*.spec.ts'",
+    "unit-test:node": "dev-tool run test:node-ts-input -- --timeout 100000 './test/internal/unit/**/*.spec.ts'",
     "update-snippets": "echo skipped"
   },
   "repository": "github:Azure/azure-sdk-for-js",

sdk/cosmosdb/cosmos/src/client/ChangeFeed/ChangeFeedForEpkRange.ts

@@ -445,23 +445,22 @@ export class ChangeFeedForEpkRange<T> implements ChangeFeedPullModelIterator<T>
         getEmptyCosmosDiagnostics(),
       );
     } catch (err) {
-      if (err.code >= StatusCodes.BadRequest && err.code !== StatusCodes.Gone) {
-        const errorResponse = new ErrorResponse(err.message);
-        errorResponse.code = err.code;
-        errorResponse.headers = err.headers;
-
-        throw errorResponse;
+      // If partition split/merge is encountered, handle it gracefully and continue fetching results.
+      if (err.code === StatusCodes.Gone) {
+        return new ChangeFeedIteratorResponse(
+          [],
+          0,
+          err.code,
+          err.headers,
+          getEmptyCosmosDiagnostics(),
+          err.substatus,
+        );
       }
-
-      // If any other errors are encountered, eg. partition split or gone, handle it based on error code and not break the flow.
-      return new ChangeFeedIteratorResponse(
-        [],
-        0,
-        err.code,
-        err.headers,
-        getEmptyCosmosDiagnostics(),
-        err.substatus,
-      );
+      // If any other errors are encountered, throw the error.
+      const errorResponse = new ErrorResponse(err.message);
+      errorResponse.code = err.code;
+      errorResponse.headers = err.headers;
+      throw errorResponse;
     }
   }
 }

sdk/cosmosdb/cosmos/src/client/ChangeFeed/ChangeFeedForPartitionKey.ts

@@ -4,7 +4,7 @@ import type { InternalChangeFeedIteratorOptions } from "./InternalChangeFeedOpti
 import { ChangeFeedIteratorResponse } from "./ChangeFeedIteratorResponse";
 import type { Container, Resource } from "../../client";
 import type { ClientContext } from "../../ClientContext";
-import { Constants, ResourceType, StatusCodes } from "../../common";
+import { Constants, ResourceType } from "../../common";
 import type { FeedOptions, Response } from "../../request";
 import { ErrorResponse } from "../../request";
 import { ContinuationTokenForPartitionKey } from "./ContinuationTokenForPartitionKey";
@@ -177,19 +177,11 @@ export class ChangeFeedForPartitionKey<T> implements ChangeFeedPullModelIterator
         getEmptyCosmosDiagnostics(),
       );
     } catch (err) {
-      if (err.code >= StatusCodes.BadRequest && err.code !== StatusCodes.Gone) {
-        const errorResponse = new ErrorResponse(err.message);
-        errorResponse.code = err.code;
-        errorResponse.headers = err.headers;
-        throw errorResponse;
-      }
-      return new ChangeFeedIteratorResponse(
-        [],
-        0,
-        err.code,
-        err.headers,
-        getEmptyCosmosDiagnostics(),
-      );
+      // If any errors are encountered, throw the error.
+      const errorResponse = new ErrorResponse(err.message);
+      errorResponse.code = err.code;
+      errorResponse.headers = err.headers;
+      throw errorResponse;
     }
   }
 }

sdk/cosmosdb/cosmos/test/internal/unit/changeFeed/ChangeFeedForEpkRange.spec.ts

@@ -0,0 +1,86 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+import assert from "assert";
+import sinon from "sinon";
+import { ChangeFeedForEpkRange, ChangeFeedMode } from "../../../../src/client/ChangeFeed";
+import { ClientContext, Container, ErrorResponse, TimeoutError } from "../../../../src";
+import { PartitionKeyRangeCache, QueryRange } from "../../../../src/routing";
+import { ChangeFeedRange } from "../../../../src/client/ChangeFeed/ChangeFeedRange";
+import { MockedQueryIterator } from "../../../public/common/MockQueryIterator";
+
+interface Resource {
+  id: string;
+  [key: string]: any;
+}
+
+class MockClientContext {
+  constructor(private partitionKeyRanges: unknown) {}
+  public queryPartitionKeyRanges(): MockedQueryIterator {
+    return new MockedQueryIterator(this.partitionKeyRanges);
+  }
+  public queryFeed() {
+    throw new TimeoutError();
+  }
+  public getClientConfig() {
+    return {};
+  }
+  public recordDiagnostics() {
+    return;
+  }
+}
+
+describe("ChangeFeedForEpkRange Unit Tests", () => {
+  const partitionKeyRanges = [{ id: "0", minInclusive: "", maxExclusive: "FF" }];
+
+  const clientContext: ClientContext = new MockClientContext(partitionKeyRanges) as any;
+  const partitionKeyRangeCache = new PartitionKeyRangeCache(clientContext);
+  let container: sinon.SinonStubbedInstance<Container>;
+  let changeFeedForEpkRange: ChangeFeedForEpkRange<Resource>;
+
+  beforeEach(() => {
+    container = sinon.createStubInstance(Container);
+
+    // Initialize ChangeFeedForEpkRange instance with mocked dependencies
+    changeFeedForEpkRange = new ChangeFeedForEpkRange<Resource>(
+      clientContext,
+      container as unknown as Container,
+      partitionKeyRangeCache as PartitionKeyRangeCache,
+      "resource-id",
+      "/dbs/db/colls/coll",
+      "https://localhost:8081/dbs/db/colls/coll",
+      {
+        continuationToken: undefined,
+        startFromNow: false,
+        startTime: undefined,
+        changeFeedMode: ChangeFeedMode.LatestVersion,
+        maxItemCount: 10,
+        sessionToken: "session-token",
+      },
+      new QueryRange("", "FF", true, false),
+    );
+
+    sinon.stub(changeFeedForEpkRange as any, "setIteratorRid").resolves();
+    sinon.stub(changeFeedForEpkRange as any, "fillChangeFeedQueue").resolves();
+    (changeFeedForEpkRange as any).isInstantiated = true;
+
+    const mockFeedRange = new ChangeFeedRange("", "FF", "abc");
+    (changeFeedForEpkRange as any).queue.enqueue(mockFeedRange);
+  });
+
+  afterEach(() => {
+    sinon.restore();
+  });
+
+  it("should throw TimeOutError instead of handling it internally", async () => {
+    const timeOutError = new TimeoutError();
+
+    try {
+      await changeFeedForEpkRange.readNext();
+      assert.fail("should throw exception");
+    } catch (err) {
+      assert(err instanceof ErrorResponse);
+      assert.strictEqual(err.message, "Timeout Error");
+      assert.strictEqual(err.code, timeOutError.code);
+    }
+  });
+});