Expose retries

cloud-agnostic/core/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@itwin/cloud-agnostic-core",
-  "version": "3.0.5",
+  "version": "3.1.0",
   "description": "Package that allows configuring components loaded by dependency injection",
   "keywords": [
     "Bentley",

common/config/rush/version-policies.json

@@ -2,7 +2,7 @@
   {
     "definitionName": "lockStepVersion",
     "policyName": "lockStepVersionObjectStorage",
-    "version": "3.0.5",
+    "version": "3.1.0",
     "nextBump": "prerelease"
   }
 ]

storage/azure/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@itwin/object-storage-azure",
-  "version": "3.0.5",
+  "version": "3.1.0",
   "description": "Object storage implementation using Azure Blob Storage",
   "keywords": [
     "Bentley",

storage/azure/src/server/AzureServerStorage.ts

@@ -29,6 +29,7 @@ import {
   ObjectReference,
   EntityPageListIterator,
   ServerStorage,
+  StorageOptions,
   TransferData,
   TransferType,
 } from "@itwin/object-storage-core";
@@ -42,7 +43,7 @@ import {
 } from "./internal";
 import { BlobServiceClientWrapper, BlockBlobClientWrapper } from "./wrappers";
 
-export interface AzureServerStorageConfig {
+export interface AzureServerStorageConfig extends StorageOptions {
   accountName: string;
   accountKey: string;
   baseUrl: string;
@@ -144,7 +145,9 @@ export class AzureServerStorage extends ServerStorage {
   ): EntityPageListIterator<ObjectReference> {
     const pageIterator: EntityPageListIterator<ObjectReference> =
       new EntityPageListIterator(() =>
-        this._client.getObjectsNextPage(directory, { maxPageSize: maxPageSize })
+        this._client.getObjectsNextPage(directory, {
+          maxPageSize: maxPageSize,
+        })
       );
     return pageIterator;
   }

storage/azure/src/server/AzureServerStorageBindings.ts

@@ -2,11 +2,6 @@
  * Copyright (c) Bentley Systems, Incorporated. All rights reserved.
  * See LICENSE.md in the project root for license terms and full copyright notice.
  *--------------------------------------------------------------------------------------------*/
-import {
-  BlobServiceClient,
-  StorageSharedKeyCredential,
-} from "@azure/storage-blob";
-
 import { ConfigError } from "@itwin/cloud-agnostic-core/lib/internal";
 
 import { DependencyConfig, DIContainer } from "@itwin/cloud-agnostic-core";
@@ -22,7 +17,10 @@ import {
   AzureServerStorage,
   AzureServerStorageConfig,
 } from "./AzureServerStorage";
-import { BlobServiceClientWrapper } from "./wrappers/BlobServiceClientWrapper";
+import {
+  BlobServiceClientWrapper,
+  BlobServiceClientWrapperFactory,
+} from "./wrappers";
 
 export type AzureServerStorageBindingsConfig = AzureServerStorageConfig &
   DependencyConfig;
@@ -55,20 +53,20 @@ export class AzureServerStorageBindings extends ServerStorageDependency {
       }
     );
 
+    container.registerFactory(
+      BlobServiceClientWrapperFactory,
+      (c: DIContainer) => {
+        const resolvedConfig = c.resolve<AzureServerStorageBindingsConfig>(
+          Types.AzureServer.config
+        );
+        return new BlobServiceClientWrapperFactory(resolvedConfig.retryOptions);
+      }
+    );
     container.registerFactory(BlobServiceClientWrapper, (c: DIContainer) => {
-      return new BlobServiceClientWrapper(c.resolve(BlobServiceClient));
-    });
-    container.registerFactory(BlobServiceClient, (c: DIContainer) => {
       const resolvedConfig = c.resolve<AzureServerStorageBindingsConfig>(
         Types.AzureServer.config
       );
-      return new BlobServiceClient(
-        resolvedConfig.baseUrl,
-        new StorageSharedKeyCredential(
-          resolvedConfig.accountName,
-          resolvedConfig.accountKey
-        )
-      );
+      return c.resolve(BlobServiceClientWrapperFactory).create(resolvedConfig);
     });
   }
 }

storage/azure/src/server/wrappers/BlobServiceClientWrapperFactory.ts

@@ -0,0 +1,47 @@
+/*---------------------------------------------------------------------------------------------
+ * Copyright (c) Bentley Systems, Incorporated. All rights reserved.
+ * See LICENSE.md in the project root for license terms and full copyright notice.
+ *--------------------------------------------------------------------------------------------*/
+import {
+  BlobServiceClient,
+  newPipeline,
+  StoragePipelineOptions,
+  StorageSharedKeyCredential,
+} from "@azure/storage-blob";
+
+import { RetryOptions } from "@itwin/object-storage-core";
+
+import { BlobServiceClientWrapper } from "./BlobServiceClientWrapper";
+
+export interface AzureBlobServiceConfig {
+  accountName: string;
+  accountKey: string;
+  baseUrl: string;
+}
+
+export class BlobServiceClientWrapperFactory {
+  public constructor(private readonly _retryOptions: RetryOptions = {}) {}
+
+  public create(config: AzureBlobServiceConfig): BlobServiceClientWrapper {
+    const credential = new StorageSharedKeyCredential(
+      config.accountName,
+      config.accountKey
+    );
+    const pipelineOptions: StoragePipelineOptions = {
+      retryOptions: {
+        maxTries:
+          this._retryOptions.maxRetries != undefined
+            ? this._retryOptions.maxRetries + 1
+            : undefined,
+        retryDelayInMs: this._retryOptions.retryDelayMs,
+        maxRetryDelayInMs: this._retryOptions.maxRetryDelayMs,
+      },
+    };
+    return new BlobServiceClientWrapper(
+      new BlobServiceClient(
+        config.baseUrl,
+        newPipeline(credential, pipelineOptions)
+      )
+    );
+  }
+}

storage/azure/src/server/wrappers/index.ts

@@ -3,5 +3,6 @@
  * See LICENSE.md in the project root for license terms and full copyright notice.
  *--------------------------------------------------------------------------------------------*/
 export * from "./BlobServiceClientWrapper";
+export * from "./BlobServiceClientWrapperFactory";
 export * from "./BlockBlobClientWrapper";
 export * from "./BlockBlobClientWrapperFactory";

storage/core/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@itwin/object-storage-core",
-  "version": "3.0.5",
+  "version": "3.1.0",
   "description": "Core generic object storage interfaces",
   "keywords": [
     "Bentley",

storage/core/src/client/ClientStorage.ts

@@ -14,11 +14,15 @@ import {
 
 export abstract class ClientStorage {
   public abstract download(
-    input: (UrlDownloadInput | ConfigDownloadInput) & { transferType: "buffer" }
+    input: (UrlDownloadInput | ConfigDownloadInput) & {
+      transferType: "buffer";
+    }
   ): Promise<Buffer>;
 
   public abstract download(
-    input: (UrlDownloadInput | ConfigDownloadInput) & { transferType: "stream" }
+    input: (UrlDownloadInput | ConfigDownloadInput) & {
+      transferType: "stream";
+    }
   ): Promise<Readable>;
 
   public abstract download(

storage/core/src/common/Interfaces.ts

@@ -54,3 +54,25 @@ export interface ConfigTransferInput {
   reference: ObjectReference;
   transferConfig: TransferConfig;
 }
+
+export interface RetryOptions {
+  /** Maximum number of retry attempts. */
+  maxRetries?: number;
+  /**
+   * Initial delay in milliseconds before the first retry, with exponential
+   * backoff applied on subsequent attempts.
+   * Supported by Azure server storage and all URL-based client transfers.
+   * Ignored by S3, Minio, and Google storage.
+   */
+  retryDelayMs?: number;
+  /**
+   * Upper bound in milliseconds for the delay between retries.
+   * Supported by Azure and Google storage, and all URL-based client transfers.
+   * Ignored by S3 and Minio server storage.
+   */
+  maxRetryDelayMs?: number;
+}
+
+export interface StorageOptions {
+  retryOptions?: RetryOptions;
+}

storage/core/src/server/internal/Helpers.ts

@@ -3,17 +3,11 @@
  * See LICENSE.md in the project root for license terms and full copyright notice.
  *--------------------------------------------------------------------------------------------*/
 import { randomBytes } from "crypto";
-import { createReadStream, createWriteStream, promises } from "fs";
+import { createWriteStream, promises } from "fs";
 import { dirname } from "path";
 import { Readable } from "stream";
 
-import axios from "axios";
-
-import {
-  ConfigTransferInput,
-  createClientAbortSignal,
-  UrlTransferInput,
-} from "../../common";
+import { ConfigTransferInput, UrlTransferInput } from "../../common";
 import { defaultExpiresInSeconds } from "../../common/internal";
 import {
   ConfigDownloadInput,
@@ -55,7 +49,11 @@ export async function streamToBuffer(stream: Readable): Promise<Buffer> {
       // eslint-disable-next-line @typescript-eslint/no-unsafe-argument
       chunks.push(data instanceof Buffer ? data : Buffer.from(data))
     );
-    stream.on("end", () => resolve(Buffer.concat(chunks)));
+    stream.on("end", () => {
+      const buffer = Buffer.concat(chunks);
+      stream.destroy();
+      resolve(buffer);
+    });
     stream.on("error", reject);
   });
 }
@@ -122,104 +120,11 @@ export async function streamToTransferType(
   }
 }
 
-export async function downloadFromUrl(
-  input: UrlDownloadInput
-): Promise<TransferData> {
-  const { transferType, url, abortSignal } = input;
-
-  // There is an issue with Axios type definitions. Casting should be removed
-  // after upgrading to Axios 1.0
-  // See: https://github.com/axios/axios/pull/4229
-  const signal = abortSignal ? createClientAbortSignal(abortSignal) : undefined;
-
-  switch (transferType) {
-    case "buffer":
-      return downloadFromUrlAsBuffer(url, signal);
-    case "stream":
-      return downloadFromUrlAsStream(url, signal);
-    case "local":
-      const localPath = input.localPath;
-      assertLocalFile(localPath);
-      return downloadFromUrlToLocalFile(url, localPath, signal);
-    default:
-      throw new Error(`Type ${input.transferType} is not supported`);
-  }
-}
-
-export async function uploadToUrl(
-  url: string,
-  data: TransferData,
-  headers?: Record<string, string>
-): Promise<void> {
-  let dataToUpload: Readable | Buffer;
-  if (typeof data === "string") {
-    await assertFileNotEmpty(data);
-    dataToUpload = createReadStream(data);
-  } else {
-    dataToUpload = data;
-  }
-  await axios.put(url, dataToUpload, {
-    headers,
-  });
-}
-
 // TODO: switch to using crypto.randomUUID function once support for Node 12.x is dropped.
 export function getRandomString(): string {
   return randomBytes(16).toString("hex");
 }
 
-async function downloadFromUrlAsBuffer(
-  url: string,
-  signal?: AbortSignal
-): Promise<Buffer> {
-  let promise = axios.get(url, {
-    responseType: "arraybuffer",
-    signal,
-  });
-  promise = convertAbortErrorName(promise);
-  return (await promise).data as Buffer;
-}
-
-async function downloadFromUrlAsStream(
-  url: string,
-  signal?: AbortSignal
-): Promise<Readable> {
-  let promise = axios.get(url, {
-    responseType: "stream",
-    signal,
-  });
-  promise = convertAbortErrorName(promise);
-  return (await promise).data as Readable;
-}
-
-async function downloadFromUrlToLocalFile(
-  url: string,
-  localPath: string,
-  signal?: AbortSignal
-): Promise<string> {
-  let promise = axios.get(url, {
-    responseType: "stream",
-    signal,
-  });
-  promise = convertAbortErrorName(promise);
-
-  const stream = (await promise).data as Readable;
-  await streamToLocalFile(stream, localPath);
-
-  return localPath;
-}
-
-async function convertAbortErrorName<T>(promise: Promise<T>): Promise<T> {
-  try {
-    return await promise;
-  } catch (error: unknown) {
-    if (error instanceof Error && error.name === "CanceledError")
-      error.name = "AbortError";
-
-    throw error;
-  }
-}
-
 export function getExpiryDate(options?: ExpiryOptions): Date {
   if (options?.expiresInSeconds && options?.expiresOn) {
     throw new Error(