Storage adapter: Propagate bulk errors (#242901)

Currently, the `bulk` method of the storage adapter is not failing if
one of the operations fails. This can hide problems, e.g. if saving a
stream would cause the `create` operation in the bulk call to fail, it
would fail silently and tell the user saving worked fine.

With the change on this PR, the error is passed back to the user
correctly (had to mess with the mappings of the index to provoke an
error, not sure whether it can be triggered accidentally):
<img width="309" height="492" alt="Screenshot 2025-11-13 at 14 08 36"
src="https://github.com/user-attachments/assets/aaba6185-23e3-475f-9545-fd595b5557e2"
/>

This is a breaking change of the storage adapter interface, but judging
from the current usages we don't want this kind of error silently being
ignored - throwing an error makes sure the problem is visible.

Author: flash1293

src/platform/packages/shared/kbn-storage-adapter/index.ts

@@ -55,12 +55,21 @@ export type StorageClientBulkOperation<TDocument extends { _id?: string }> =
     }
   | { delete: { _id: string } };
 
+export interface StorageClientBulkOptions {
+  /**
+   * If true, throws BulkOperationError when any operation in the bulk request fails.
+   * If false (default), returns the response with errors field populated, similar to Promise.allSettled behavior.
+   * @default false
+   */
+  throwOnFail?: boolean;
+}
+
 export type StorageClientBulkRequest<TDocument extends { _id?: string }> = Omit<
   BulkRequest,
   'operations' | 'index'
 > & {
   operations: Array<StorageClientBulkOperation<TDocument>>;
-};
+} & StorageClientBulkOptions;
 export type StorageClientBulkResponse = BulkResponse;
 
 export type StorageClientDeleteRequest = Omit<DeleteRequest, 'index'>;
@@ -91,6 +100,13 @@ export type StorageClientSearch<TDocumentType = never> = <
   request: TSearchRequest
 ) => Promise<StorageClientSearchResponse<TDocumentType, TSearchRequest>>;
 
+/**
+ * Performs bulk operations on documents.
+ *
+ * By default, behaves similar to Promise.allSettled - individual operation failures
+ * are returned in the response without throwing an error. Set `throwOnFail: true`
+ * to throw a BulkOperationError when any operation fails.
+ */
 export type StorageClientBulk<TDocumentType extends { _id?: string } = never> = (
   request: StorageClientBulkRequest<TDocumentType>
 ) => Promise<StorageClientBulkResponse>;
@@ -159,4 +175,6 @@ export type StorageDocumentOf<TStorageSettings extends StorageSettings> = Partia
 
 export { StorageIndexAdapter } from './src/index_adapter';
 
+export { BulkOperationError } from './src/errors';
+
 export { types } from './types';

src/platform/packages/shared/kbn-storage-adapter/src/errors.ts

@@ -0,0 +1,17 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the "Elastic License
+ * 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
+ * Public License v 1"; you may not use this file except in compliance with, at
+ * your election, the "Elastic License 2.0", the "GNU Affero General Public
+ * License v3.0 only", or the "Server Side Public License, v 1".
+ */
+
+import type { BulkResponse } from '@elastic/elasticsearch/lib/api/types';
+
+export class BulkOperationError extends Error {
+  constructor(message: string, public response: BulkResponse) {
+    super(message);
+    this.name = 'BulkOperationError';
+  }
+}

src/platform/packages/shared/kbn-storage-adapter/src/index_adapter/index.ts

@@ -9,6 +9,7 @@
 
 import type {
   BulkOperationContainer,
+  BulkOperationType,
   IndexResponse,
   IndicesIndexState,
   IndicesIndexTemplate,
@@ -40,6 +41,7 @@ import type {
 } from '../..';
 import { getSchemaVersion } from '../get_schema_version';
 import type { StorageMappingProperty } from '../../types';
+import { BulkOperationError } from '../errors';
 
 function getAliasName(name: string) {
   return name;
@@ -351,6 +353,7 @@ export class StorageIndexAdapter<
   private bulk: StorageClientBulk<TApplicationType> = ({
     operations,
     refresh = 'wait_for',
+    throwOnFail = false,
     ...request
   }): Promise<StorageClientBulkResponse> => {
     if (operations.length === 0) {
@@ -393,6 +396,21 @@ export class StorageIndexAdapter<
     };
 
     return this.validateComponentsBeforeWriting(attemptBulk).then(async (response) => {
+      // Check for errors and throw if throwOnFail is true
+      if (throwOnFail) {
+        const erroredItems = response.items.filter((item) => {
+          const operation = Object.keys(item)[0] as BulkOperationType;
+          return item[operation]?.error;
+        });
+        if (erroredItems.length > 0) {
+          throw new BulkOperationError(
+            `Bulk operation failed for ${erroredItems.length} out of ${
+              response.items.length
+            } items: ${JSON.stringify(erroredItems)}`,
+            response
+          );
+        }
+      }
       return response;
     });
   };

src/platform/packages/shared/kbn-storage-adapter/src/index_adapter/integration_tests/index.test.ts

@@ -14,7 +14,7 @@ import type {
   StorageClientIndexResponse,
   StorageDocumentOf,
 } from '../../..';
-import { StorageIndexAdapter, type StorageSettings } from '../../..';
+import { BulkOperationError, StorageIndexAdapter, type StorageSettings } from '../../..';
 import type { Logger } from '@kbn/core/server';
 import * as getSchemaVersionModule from '../../get_schema_version';
 import { isResponseError } from '@kbn/es-errors';
@@ -348,6 +348,99 @@ describe('StorageIndexAdapter', () => {
     });
   });
 
+  describe('when bulk operation encounters errors', () => {
+    afterAll(async () => {
+      await client?.clean();
+    });
+
+    it('throws BulkOperationError when bulk operation contains document-level errors', async () => {
+      // Create an adapter with strict mapping to trigger mapping errors
+      const strictAdapter = createStorageIndexAdapter({
+        name: 'test_strict_index',
+        schema: {
+          properties: {
+            foo: {
+              type: 'keyword',
+            },
+          },
+        },
+      });
+      const strictClient = strictAdapter.getClient();
+
+      // First create the index with strict mapping
+      await strictClient.index({
+        id: 'doc1',
+        document: { foo: 'bar' },
+      });
+
+      // Try to bulk index with an invalid field (should fail due to dynamic: strict)
+      await expect(
+        strictClient.bulk({
+          operations: [
+            {
+              index: {
+                _id: 'doc2',
+                document: { foo: 'baz', invalid_field: 'value' } as any,
+              },
+            },
+          ],
+          refresh: 'wait_for',
+          throwOnFail: true,
+        })
+      ).rejects.toThrow(BulkOperationError);
+
+      await strictClient.clean();
+    });
+
+    it('includes error details in the thrown BulkOperationError', async () => {
+      const strictAdapter = createStorageIndexAdapter({
+        name: 'test_strict_index_2',
+        schema: {
+          properties: {
+            foo: {
+              type: 'keyword',
+            },
+          },
+        },
+      });
+      const strictClient = strictAdapter.getClient();
+
+      // Create the index
+      await strictClient.index({
+        id: 'doc1',
+        document: { foo: 'bar' },
+      });
+
+      try {
+        await strictClient.bulk({
+          operations: [
+            {
+              index: {
+                _id: 'doc2',
+                document: { foo: 'baz', invalid_field: 'value' } as any,
+              },
+            },
+          ],
+          throwOnFail: true,
+          refresh: 'wait_for',
+        });
+        fail('Expected BulkOperationError to be thrown');
+      } catch (err) {
+        const error = err as BulkOperationError;
+        expect(error).toBeInstanceOf(BulkOperationError);
+        expect(error.message).toContain('Bulk operation failed');
+        expect(error.message).toContain('1 out of 1 items');
+        expect(error.response).toBeDefined();
+        expect(error.response.errors).toBe(true);
+        expect(error.response.items).toHaveLength(1);
+        expect(error.response.items[0].index?.error).toBeDefined();
+        expect(error.response.items[0].index?.error?.type).toBe('strict_dynamic_mapping_exception');
+      }
+
+      await strictClient.clean();
+    });
+  });
+
   describe('when writing/bootstrapping with an existing, incompatible index', () => {
     beforeAll(async () => {
       await client.index({ id: 'foo', document: { foo: 'bar' } });

x-pack/platform/plugins/shared/streams/server/lib/streams/assets/asset_client.ts

@@ -279,6 +279,7 @@ export class AssetClient {
           },
         };
       }),
+      throwOnFail: true,
     });
   }
 

x-pack/platform/plugins/shared/streams/server/lib/streams/attachments/attachment_client.ts

@@ -386,6 +386,7 @@ export class AttachmentClient {
 
     const bulkResponse = await this.clients.storageClient.bulk({
       operations: bulkOperations,
+      throwOnFail: true,
     });
 
     return { errors: bulkResponse.errors, items: bulkResponse.items };

x-pack/platform/plugins/shared/streams/server/lib/streams/feature/feature_client.ts

@@ -150,6 +150,7 @@ export class FeatureClient {
           },
         };
       }),
+      throwOnFail: true,
     });
   }
 

x-pack/platform/plugins/shared/streams/server/lib/streams/state_management/execution_plan/execution_plan.ts

@@ -424,6 +424,7 @@ export class ExecutionPlan {
     return this.dependencies.storageClient.bulk({
       operations: actions.map(dotDocumentActionToBulkOperation),
       refresh: true,
+      throwOnFail: true,
     });
   }