From 9dd8c711ebfc226ec60ddf80facce3df8bca1a0a Mon Sep 17 00:00:00 2001 From: kubasobon Date: Thu, 10 Sep 2026 16:44:43 +0200 Subject: [PATCH 1/8] entity_store: SO v9, nonPriority cursor field --- .../entity-engine-descriptor-v2/10.9.0.json | 102 ++++++++++++++++++ .../engine_descriptor/constants.ts | 3 + .../saved_objects/engine_descriptor/types.ts | 15 +++ 3 files changed, 120 insertions(+) create mode 100644 packages/kbn-check-saved-objects-cli/src/migrations/__fixtures__/entity-engine-descriptor-v2/10.9.0.json diff --git a/packages/kbn-check-saved-objects-cli/src/migrations/__fixtures__/entity-engine-descriptor-v2/10.9.0.json b/packages/kbn-check-saved-objects-cli/src/migrations/__fixtures__/entity-engine-descriptor-v2/10.9.0.json new file mode 100644 index 0000000000000..6327bd2e18384 --- /dev/null +++ b/packages/kbn-check-saved-objects-cli/src/migrations/__fixtures__/entity-engine-descriptor-v2/10.9.0.json @@ -0,0 +1,102 @@ +{ + "10.8.0": [ + { + "type": "user", + "status": "started", + "logExtractionState": { + "checkpointTimestamp": "2026-03-04T18:00:00.000Z", + "paginationId": null, + "lastExecutionTimestamp": "2026-03-04T17:37:20.000Z", + "sliceEndTimestamp": null + }, + "error": null, + "versionState": { + "version": "2", + "state": "running", + "isMigratedFromV1": true + } + }, + { + "type": "service", + "status": "stopped", + "logExtractionState": { + "checkpointTimestamp": null, + "paginationId": null, + "lastExecutionTimestamp": null, + "sliceEndTimestamp": null + }, + "error": null, + "versionState": { + "version": "2", + "state": "running", + "isMigratedFromV1": false + } + }, + { + "type": "host", + "status": "started", + "logExtractionState": { + "checkpointTimestamp": "2026-03-04T17:00:00.000Z", + "paginationId": "some-pagination-id", + "lastExecutionTimestamp": "2026-03-04T16:37:20.000Z", + "sliceEndTimestamp": "2026-03-04T17:30:00.000Z" + }, + "error": null, + "versionState": { + "version": "2", + "state": "running", + "isMigratedFromV1": false + } + } + ], + "10.9.0": [ + { + "type": "user", + "status": "started", + "logExtractionState": { + "checkpointTimestamp": "2026-03-04T18:00:00.000Z", + "paginationId": null, + "lastExecutionTimestamp": "2026-03-04T17:37:20.000Z", + "sliceEndTimestamp": null + }, + "error": null, + "versionState": { + "version": "2", + "state": "running", + "isMigratedFromV1": true + } + }, + { + "type": "service", + "status": "stopped", + "logExtractionState": { + "checkpointTimestamp": null, + "paginationId": null, + "lastExecutionTimestamp": null, + "sliceEndTimestamp": null + }, + "error": null, + "versionState": { + "version": "2", + "state": "running", + "isMigratedFromV1": false + } + }, + { + "type": "host", + "status": "started", + "logExtractionState": { + "checkpointTimestamp": "2026-03-04T17:00:00.000Z", + "paginationId": "some-pagination-id", + "lastExecutionTimestamp": "2026-03-04T16:37:20.000Z", + "sliceEndTimestamp": "2026-03-04T17:30:00.000Z" + }, + "error": null, + "versionState": { + "version": "2", + "state": "running", + "isMigratedFromV1": false + } + } + ] +} diff --git a/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/constants.ts b/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/constants.ts index 00c876aedda8e..e9784b66c5285 100644 --- a/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/constants.ts +++ b/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/constants.ts @@ -45,6 +45,9 @@ export const EngineDescriptor = z.object({ logExtractionState: EngineLogExtractionState, /** Per entity-type log extraction overrides. Optional: descriptors written before model version 8 do not have the field. */ logExtractionConfig: LogExtractionTypeOverride.optional(), + /** Non-priority process cursor. Absent on descriptors written before model version 9, and null + * while dual-process is off. Both mean "no cursor": extraction starts from now - lookbackPeriod. */ + nonPriorityLogExtractionState: EngineLogExtractionState.nullish(), error: EngineError.nullable().default(null), versionState: VersionState, }); diff --git a/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.ts b/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.ts index 098b39b5febc0..f93da3af9e8a4 100644 --- a/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.ts +++ b/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.ts @@ -381,6 +381,20 @@ const version8: SavedObjectsFullModelVersion = { }, }; +const engineDescriptorSchemaV9 = engineDescriptorSchemaV8.extends({ + nonPriorityLogExtractionState: schema.nullable(logExtractionRuntimeStateSchemaV7), +}); + +// Adds the non-priority process cursor. Nullable with a null default, so descriptors written by +// version 8 need no backfill. Not queried, so no mappings addition. +const version9: SavedObjectsFullModelVersion = { + changes: [], + schemas: { + create: engineDescriptorSchemaV9, + forwardCompatibility: engineDescriptorSchemaV9.extends({}, { unknowns: 'ignore' }), + }, +}; + export const EngineDescriptorType: SavedObjectsType = { name: EngineDescriptorTypeName, hidden: false, @@ -395,6 +409,7 @@ export const EngineDescriptorType: SavedObjectsType = { 6: version6, 7: version7, 8: version8, + 9: version9, }, hiddenFromHttpApis: true, }; From 8f072089dc7887535995f0442e36e9eaf759a2c5 Mon Sep 17 00:00:00 2001 From: kubasobon Date: Thu, 10 Sep 2026 16:45:21 +0200 Subject: [PATCH 2/8] entity_store: route cursor by extraction mode --- .../logs_extraction/logs_extraction_client.ts | 23 ++++++++++++++----- 1 file changed, 17 insertions(+), 6 deletions(-) diff --git a/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts b/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts index fcbd0f2b6b094..126ad6f23b708 100644 --- a/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts +++ b/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts @@ -52,6 +52,7 @@ import { } from '../asset_manager/external_indices_contants'; import { type LogExtractionConfig } from '../saved_objects'; import { + type EngineDescriptor, type EngineDescriptorClient, type EngineLogExtractionState, type EntityStoreGlobalStateClient, @@ -132,6 +133,14 @@ export class LogsExtractionClient { this.extractionMode = extractionMode ?? 'single'; } + /** Cursor patch for this client's extraction mode. The non-priority process keeps its own cursor + * so the two processes cannot clobber each other's resume point. */ + private cursorPatch(state: EngineLogExtractionState): Partial { + return this.extractionMode === 'nonPriority' + ? { nonPriorityLogExtractionState: state } + : { logExtractionState: state }; + } + private async getLogExtractionConfigAndState( type: EntityType ): Promise<{ config: LogExtractionConfig; engineState: EngineLogExtractionState }> { @@ -140,9 +149,13 @@ export class LogsExtractionClient { throw new EntityStoreNotRunningError(); } const globalOverrides = await this.globalStateClient.findLogExtractionOverrides(); + const engineState = + this.extractionMode === 'nonPriority' + ? (engineDescriptor.nonPriorityLogExtractionState ?? FRESH_ENGINE_LOG_EXTRACTION_STATE) + : engineDescriptor.logExtractionState; return { config: getMergedConfig(type, globalOverrides, engineDescriptor.logExtractionConfig), - engineState: engineDescriptor.logExtractionState, + engineState, }; } @@ -206,12 +219,12 @@ export class LogsExtractionClient { await this.engineDescriptorClient.update(type, { error: null }); } else { await this.engineDescriptorClient.update(type, { - logExtractionState: { + ...this.cursorPatch({ checkpointTimestamp: null, paginationId: null, lastExecutionTimestamp: lastSearchTimestamp || moment().utc().toISOString(), sliceEndTimestamp: null, - }, + }), error: null, }); } @@ -957,9 +970,7 @@ export class LogsExtractionClient { if (opts?.specificWindow) { return; } - await this.engineDescriptorClient.update(type, { - logExtractionState: logExtractionState as EngineLogExtractionState, - }); + await this.engineDescriptorClient.update(type, this.cursorPatch(logExtractionState as EngineLogExtractionState)); } private async handleError( From b1d4edc34908edf0f498a46b5944763ae8f98c1a Mon Sep 17 00:00:00 2001 From: kubasobon Date: Thu, 10 Sep 2026 16:47:01 +0200 Subject: [PATCH 3/8] entity_store: tests for nonPriority cursor routing --- .../logs_extraction_client.test.ts | 122 ++++++++++++++++++ .../engine_descriptor/types.test.ts | 103 +++++++++++++++ 2 files changed, 225 insertions(+) create mode 100644 x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.test.ts diff --git a/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.test.ts b/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.test.ts index a70760a580ec6..1700bae117093 100644 --- a/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.test.ts +++ b/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.test.ts @@ -94,6 +94,12 @@ function createMockEngineDescriptor( paginationId: string; lastExecutionTimestamp: string; sliceEndTimestamp: string; + nonPriorityLogExtractionState: { + checkpointTimestamp: string | null; + paginationId: string | null; + lastExecutionTimestamp: string | null; + sliceEndTimestamp: string | null; + } | null; }> ) { const logExtractionState = { @@ -106,6 +112,7 @@ function createMockEngineDescriptor( type, status: ENGINE_STATUS.STARTED, logExtractionState, + nonPriorityLogExtractionState: overrides?.nonPriorityLogExtractionState ?? null, versionState: { version: 2, state: 'running' as const, isMigratedFromV1: false }, }; } @@ -2026,3 +2033,118 @@ describe('LogsExtractionClient mid-slice resume', () => { } ); }); + +describe('LogsExtractionClient extraction mode cursor routing', () => { + const fixedNow = new Date('2025-01-15T12:00:00.000Z'); + + const extractionColumns: ESQLSearchResponse['columns'] = [ + { name: '@timestamp', type: 'date' }, + { name: HASHED_ID_FIELD, type: 'keyword' }, + { name: ENGINE_METADATA_UNTYPED_ID_FIELD, type: 'keyword' }, + ]; + + function createContextWithMode(mode: 'single' | 'priority' | 'nonPriority') { + jest.clearAllMocks(); + mockExecuteEsqlQuery.mockReset(); + mockIngestEntities.mockReset(); + + const mockLogger = loggerMock.create(); + const mockEsClient = { + indices: { + resolveIndex: jest.fn().mockResolvedValue({ indices: [], aliases: [], data_streams: [] }), + }, + } as unknown as jest.Mocked; + const mockDataViewsService = { + get: jest.fn().mockResolvedValue({ getIndexPattern: jest.fn().mockReturnValue('logs-*') }), + } as unknown as jest.Mocked; + const mockEngineDescriptorClient: jest.Mocked< + Pick + > = { + findOrThrow: jest.fn(), + update: jest.fn().mockResolvedValue({}), + }; + const mockGlobalStateClient = createMockGlobalStateClient(); + + const client = new LogsExtractionClient({ + logger: mockLogger, + namespace: 'default', + esClient: mockEsClient, + dataViewsService: mockDataViewsService, + engineDescriptorClient: mockEngineDescriptorClient as unknown as EngineDescriptorClient, + globalStateClient: mockGlobalStateClient as unknown as EntityStoreGlobalStateClient, + extractionMode: mode, + }); + + return { client, mockEngineDescriptorClient, mockDataViewsService }; + } + + beforeEach(() => { + jest.useFakeTimers({ now: fixedNow.getTime() }); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('nonPriority mode writes nonPriorityLogExtractionState on mid-run and end-of-run persists', async () => { + const { client, mockEngineDescriptorClient } = createContextWithMode('nonPriority'); + mockEngineDescriptorClient.findOrThrow.mockResolvedValue( + createMockEngineDescriptor('user') as Awaited> + ); + mockIngestEntities.mockResolvedValue(undefined); + // probe → extraction (1 row, non-final) → empty probe (end of window) → sweep + mockExecuteEsqlQuery + .mockResolvedValueOnce(mockLogPaginationCursorProbeRow('2025-01-15T11:00:00.000Z')) + .mockResolvedValueOnce({ + columns: extractionColumns, + values: [['2025-01-15T10:30:00.000Z', 'hash1', 'entity1']], + }) + .mockResolvedValueOnce(mockLogPaginationCursorProbeEmpty()) + .mockResolvedValueOnce({ columns: extractionColumns, values: [] }); + + await client.extractLogs('user'); + + const updateCalls = mockEngineDescriptorClient.update.mock.calls.map(([, update]) => update); + // Every update must use nonPriorityLogExtractionState, never logExtractionState. + expect(updateCalls.every((u) => !('logExtractionState' in u))).toBe(true); + expect(updateCalls.some((u) => 'nonPriorityLogExtractionState' in u)).toBe(true); + }); + + it('single mode writes logExtractionState — regression guard for the default path', async () => { + const { client, mockEngineDescriptorClient } = createContextWithMode('single'); + mockEngineDescriptorClient.findOrThrow.mockResolvedValue( + createMockEngineDescriptor('user') as Awaited> + ); + mockIngestEntities.mockResolvedValue(undefined); + mockExtractSuccessSequence({ columns: extractionColumns, values: [] }); + + await client.extractLogs('user'); + + const updateCalls = mockEngineDescriptorClient.update.mock.calls.map(([, update]) => update); + expect(updateCalls.every((u) => !('nonPriorityLogExtractionState' in u))).toBe(true); + }); + + it('nonPriority with a live logExtractionState checkpoint starts from lookbackPeriod, not from the priority cursor', async () => { + // The priority process has a live checkpoint; the non-priority cursor is absent (null). + // The non-priority client must not read the priority cursor — it starts fresh. + const { client, mockEngineDescriptorClient } = createContextWithMode('nonPriority'); + const priorityCheckpoint = '2025-01-14T00:00:00.000Z'; // 36 hours ago — outside lookback + mockEngineDescriptorClient.findOrThrow.mockResolvedValue( + createMockEngineDescriptor('user', { + checkpointTimestamp: priorityCheckpoint, + lastExecutionTimestamp: priorityCheckpoint, + // nonPriorityLogExtractionState absent → null default → fresh start + }) as Awaited> + ); + mockIngestEntities.mockResolvedValue(undefined); + mockExtractSuccessSequence({ columns: extractionColumns, values: [] }); + + await client.extractLogs('user'); + + // The first query should probe from lookbackPeriod (3h back from fixedNow = 09:00), + // not from the priority cursor 36 hours ago. + const firstQuery = mockExecuteEsqlQuery.mock.calls[0][0].query; + expect(firstQuery).not.toContain(priorityCheckpoint); + expect(firstQuery).toContain('2025-01-15T09:00:00.000Z'); + }); +}); diff --git a/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.test.ts b/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.test.ts new file mode 100644 index 0000000000000..e49c0c65970c2 --- /dev/null +++ b/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.test.ts @@ -0,0 +1,103 @@ +/* + * 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; you may not use this file except in compliance with the Elastic License + * 2.0. + */ + +import type { SavedObjectsFullModelVersion } from '@kbn/core-saved-objects-server'; +import { EngineDescriptorType } from './types'; + +const modelVersions = EngineDescriptorType.modelVersions as Record< + number, + SavedObjectsFullModelVersion +>; + +/** Minimal valid v8-era descriptor. v9 accepts it unchanged via the nullable+default field. */ +const BASE_DESCRIPTOR = { + type: 'user', + status: 'started', + logExtractionState: { + checkpointTimestamp: null, + paginationId: null, + lastExecutionTimestamp: null, + sliceEndTimestamp: null, + }, + error: null, + versionState: { version: 2, state: 'running', isMigratedFromV1: false }, +}; + +describe('EngineDescriptorType model version 9 (nonPriorityLogExtractionState)', () => { + const v9 = modelVersions[9]; + const v8 = modelVersions[8]; + + it('changes is empty — no backfill needed because the field is nullable with a null default', () => { + expect(v9.changes).toEqual([]); + }); + + it('v8-era descriptor (no nonPriorityLogExtractionState) validates against the v9 create schema', () => { + expect(() => v9.schemas?.create?.validate(BASE_DESCRIPTOR)).not.toThrow(); + }); + + it('null validates (state while dual-process is off)', () => { + expect(() => + v9.schemas?.create?.validate({ ...BASE_DESCRIPTOR, nonPriorityLogExtractionState: null }) + ).not.toThrow(); + }); + + it('full four-field non-priority cursor validates', () => { + expect(() => + v9.schemas?.create?.validate({ + ...BASE_DESCRIPTOR, + nonPriorityLogExtractionState: { + checkpointTimestamp: '2026-01-01T00:00:00.000Z', + paginationId: 'some-entity-id', + lastExecutionTimestamp: '2025-12-31T23:00:00.000Z', + sliceEndTimestamp: '2026-01-01T00:30:00.000Z', + }, + }) + ).not.toThrow(); + }); + + it('mid-run cursor (non-null checkpoint, paginationId, sliceEndTimestamp) validates', () => { + expect(() => + v9.schemas?.create?.validate({ + ...BASE_DESCRIPTOR, + nonPriorityLogExtractionState: { + checkpointTimestamp: '2026-01-01T00:00:00.000Z', + paginationId: 'some-entity-id', + lastExecutionTimestamp: null, + sliceEndTimestamp: '2026-01-01T00:30:00.000Z', + }, + }) + ).not.toThrow(); + }); + + it('rejects a wrongly typed value', () => { + expect(() => + v9.schemas?.create?.validate({ + ...BASE_DESCRIPTOR, + nonPriorityLogExtractionState: { checkpointTimestamp: 12345 }, + }) + ).toThrow(); + }); + + it('v8 create schema rejects nonPriorityLogExtractionState — the field is new in v9', () => { + expect(() => + v8.schemas?.create?.validate({ ...BASE_DESCRIPTOR, nonPriorityLogExtractionState: null }) + ).toThrow(); + }); + + it('v8 forwardCompatibility drops nonPriorityLogExtractionState written by a v9 node', () => { + const result = v8.schemas?.forwardCompatibility?.validate({ + ...BASE_DESCRIPTOR, + nonPriorityLogExtractionState: { + checkpointTimestamp: '2026-01-01T00:00:00.000Z', + paginationId: null, + lastExecutionTimestamp: null, + sliceEndTimestamp: null, + }, + }); + expect(result).not.toHaveProperty('nonPriorityLogExtractionState'); + }); +}); From 22ee8f796216a13e44190ce214584985e0b037ca Mon Sep 17 00:00:00 2001 From: kubasobon Date: Fri, 11 Sep 2026 11:23:56 +0200 Subject: [PATCH 4/8] fixup: update comments --- .../server/domain/logs_extraction/logs_extraction_client.ts | 4 ++-- .../domain/saved_objects/engine_descriptor/constants.ts | 4 ++-- .../domain/saved_objects/engine_descriptor/types.test.ts | 2 +- .../server/domain/saved_objects/engine_descriptor/types.ts | 4 ++-- 4 files changed, 7 insertions(+), 7 deletions(-) diff --git a/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts b/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts index 126ad6f23b708..ca38c9900f94a 100644 --- a/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts +++ b/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts @@ -133,8 +133,8 @@ export class LogsExtractionClient { this.extractionMode = extractionMode ?? 'single'; } - /** Cursor patch for this client's extraction mode. The non-priority process keeps its own cursor - * so the two processes cannot clobber each other's resume point. */ + /** Returns the SO update patch for this extraction mode. Each mode writes to its own cursor field + * so the two processes do not overwrite each other's position. */ private cursorPatch(state: EngineLogExtractionState): Partial { return this.extractionMode === 'nonPriority' ? { nonPriorityLogExtractionState: state } diff --git a/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/constants.ts b/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/constants.ts index e9784b66c5285..93dffa3ce48a1 100644 --- a/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/constants.ts +++ b/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/constants.ts @@ -45,8 +45,8 @@ export const EngineDescriptor = z.object({ logExtractionState: EngineLogExtractionState, /** Per entity-type log extraction overrides. Optional: descriptors written before model version 8 do not have the field. */ logExtractionConfig: LogExtractionTypeOverride.optional(), - /** Non-priority process cursor. Absent on descriptors written before model version 9, and null - * while dual-process is off. Both mean "no cursor": extraction starts from now - lookbackPeriod. */ + /** Non-priority process cursor. Absent before model version 9, null when the non-priority process + * is not running. Both mean no cursor: extraction starts from now - lookbackPeriod. */ nonPriorityLogExtractionState: EngineLogExtractionState.nullish(), error: EngineError.nullable().default(null), versionState: VersionState, diff --git a/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.test.ts b/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.test.ts index e49c0c65970c2..7951ee1c13894 100644 --- a/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.test.ts +++ b/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.test.ts @@ -13,7 +13,7 @@ const modelVersions = EngineDescriptorType.modelVersions as Record< SavedObjectsFullModelVersion >; -/** Minimal valid v8-era descriptor. v9 accepts it unchanged via the nullable+default field. */ +/** Minimal valid v8-era descriptor. nonPriorityLogExtractionState is absent; v9 defaults it to null. */ const BASE_DESCRIPTOR = { type: 'user', status: 'started', diff --git a/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.ts b/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.ts index f93da3af9e8a4..b5cd53b534708 100644 --- a/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.ts +++ b/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.ts @@ -385,8 +385,8 @@ const engineDescriptorSchemaV9 = engineDescriptorSchemaV8.extends({ nonPriorityLogExtractionState: schema.nullable(logExtractionRuntimeStateSchemaV7), }); -// Adds the non-priority process cursor. Nullable with a null default, so descriptors written by -// version 8 need no backfill. Not queried, so no mappings addition. +// Adds the non-priority process cursor. schema.nullable defaults absent keys to null, so no +// backfill is needed for version 8 descriptors. Not queried, so no mappings addition. const version9: SavedObjectsFullModelVersion = { changes: [], schemas: { From 712e96b1a544a52169e2810c84acec4c6aef3a2c Mon Sep 17 00:00:00 2001 From: kibanamachine <42973632+kibanamachine@users.noreply.github.com> Date: Fri, 11 Sep 2026 09:50:53 +0000 Subject: [PATCH 5/8] Changes from node scripts/check --- .../domain/logs_extraction/logs_extraction_client.test.ts | 8 ++++++-- .../domain/logs_extraction/logs_extraction_client.ts | 7 +++++-- 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.test.ts b/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.test.ts index 1700bae117093..3c4beb2077bcf 100644 --- a/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.test.ts +++ b/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.test.ts @@ -2089,7 +2089,9 @@ describe('LogsExtractionClient extraction mode cursor routing', () => { it('nonPriority mode writes nonPriorityLogExtractionState on mid-run and end-of-run persists', async () => { const { client, mockEngineDescriptorClient } = createContextWithMode('nonPriority'); mockEngineDescriptorClient.findOrThrow.mockResolvedValue( - createMockEngineDescriptor('user') as Awaited> + createMockEngineDescriptor('user') as Awaited< + ReturnType + > ); mockIngestEntities.mockResolvedValue(undefined); // probe → extraction (1 row, non-final) → empty probe (end of window) → sweep @@ -2113,7 +2115,9 @@ describe('LogsExtractionClient extraction mode cursor routing', () => { it('single mode writes logExtractionState — regression guard for the default path', async () => { const { client, mockEngineDescriptorClient } = createContextWithMode('single'); mockEngineDescriptorClient.findOrThrow.mockResolvedValue( - createMockEngineDescriptor('user') as Awaited> + createMockEngineDescriptor('user') as Awaited< + ReturnType + > ); mockIngestEntities.mockResolvedValue(undefined); mockExtractSuccessSequence({ columns: extractionColumns, values: [] }); diff --git a/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts b/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts index ca38c9900f94a..6aea9389c7105 100644 --- a/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts +++ b/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts @@ -151,7 +151,7 @@ export class LogsExtractionClient { const globalOverrides = await this.globalStateClient.findLogExtractionOverrides(); const engineState = this.extractionMode === 'nonPriority' - ? (engineDescriptor.nonPriorityLogExtractionState ?? FRESH_ENGINE_LOG_EXTRACTION_STATE) + ? engineDescriptor.nonPriorityLogExtractionState ?? FRESH_ENGINE_LOG_EXTRACTION_STATE : engineDescriptor.logExtractionState; return { config: getMergedConfig(type, globalOverrides, engineDescriptor.logExtractionConfig), @@ -970,7 +970,10 @@ export class LogsExtractionClient { if (opts?.specificWindow) { return; } - await this.engineDescriptorClient.update(type, this.cursorPatch(logExtractionState as EngineLogExtractionState)); + await this.engineDescriptorClient.update( + type, + this.cursorPatch(logExtractionState as EngineLogExtractionState) + ); } private async handleError( From ac5d6399a0aabf6f77b6757342a1740bbbc36eff Mon Sep 17 00:00:00 2001 From: kubasobon Date: Fri, 11 Sep 2026 12:36:41 +0200 Subject: [PATCH 6/8] fix test schemas --- .../engine_descriptor/types.test.ts | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.test.ts b/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.test.ts index 7951ee1c13894..4630c03fe8f1c 100644 --- a/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.test.ts +++ b/x-pack/platform/plugins/shared/entity_store/server/domain/saved_objects/engine_descriptor/types.test.ts @@ -5,13 +5,21 @@ * 2.0. */ -import type { SavedObjectsFullModelVersion } from '@kbn/core-saved-objects-server'; import { EngineDescriptorType } from './types'; -const modelVersions = EngineDescriptorType.modelVersions as Record< - number, - SavedObjectsFullModelVersion ->; +interface TestSchema { + validate(input: unknown): unknown; +} + +interface TestModelVersion { + changes: unknown[]; + schemas?: { + create?: TestSchema; + forwardCompatibility?: TestSchema; + }; +} + +const modelVersions = EngineDescriptorType.modelVersions as Record; /** Minimal valid v8-era descriptor. nonPriorityLogExtractionState is absent; v9 defaults it to null. */ const BASE_DESCRIPTOR = { From 61ae44432607483c243222fd10911afb9775fcec Mon Sep 17 00:00:00 2001 From: kubasobon Date: Fri, 11 Sep 2026 15:16:21 +0200 Subject: [PATCH 7/8] fix: apply review remarks --- .../logs_extraction_client.test.ts | 23 +++++++++++++++++++ .../logs_extraction/logs_extraction_client.ts | 14 +++++++---- 2 files changed, 32 insertions(+), 5 deletions(-) diff --git a/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.test.ts b/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.test.ts index 3c4beb2077bcf..91b7c32c66536 100644 --- a/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.test.ts +++ b/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.test.ts @@ -2128,6 +2128,29 @@ describe('LogsExtractionClient extraction mode cursor routing', () => { expect(updateCalls.every((u) => !('nonPriorityLogExtractionState' in u))).toBe(true); }); + it('priority mode writes logExtractionState and resumes from the existing checkpoint', async () => { + const { client, mockEngineDescriptorClient } = createContextWithMode('priority'); + const existingCheckpoint = '2025-01-15T11:30:00.000Z'; + mockEngineDescriptorClient.findOrThrow.mockResolvedValue( + createMockEngineDescriptor('user', { + checkpointTimestamp: existingCheckpoint, + lastExecutionTimestamp: existingCheckpoint, + }) as Awaited> + ); + mockIngestEntities.mockResolvedValue(undefined); + mockExtractSuccessSequence({ columns: extractionColumns, values: [] }); + + await client.extractLogs('user'); + + const updateCalls = mockEngineDescriptorClient.update.mock.calls.map(([, update]) => update); + // priority shares logExtractionState with single — never touches nonPriorityLogExtractionState. + expect(updateCalls.every((u) => !('nonPriorityLogExtractionState' in u))).toBe(true); + expect(updateCalls.some((u) => 'logExtractionState' in u)).toBe(true); + // The first ES|QL query starts from the existing checkpoint, not from lookbackPeriod. + const firstQuery = mockExecuteEsqlQuery.mock.calls[0][0].query; + expect(firstQuery).toContain(existingCheckpoint); + }); + it('nonPriority with a live logExtractionState checkpoint starts from lookbackPeriod, not from the priority cursor', async () => { // The priority process has a live checkpoint; the non-priority cursor is absent (null). // The non-priority client must not read the priority cursor — it starts fresh. diff --git a/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts b/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts index 6aea9389c7105..2f76356c9e169 100644 --- a/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts +++ b/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts @@ -133,12 +133,16 @@ export class LogsExtractionClient { this.extractionMode = extractionMode ?? 'single'; } - /** Returns the SO update patch for this extraction mode. Each mode writes to its own cursor field - * so the two processes do not overwrite each other's position. */ + /** Maps each extraction mode to its cursor field. single and priority share logExtractionState; + * nonPriority has its own field so the two processes do not overwrite each other's position. */ + private static readonly CURSOR_FIELD: Record = { + single: 'logExtractionState', + priority: 'logExtractionState', + nonPriority: 'nonPriorityLogExtractionState', + }; + private cursorPatch(state: EngineLogExtractionState): Partial { - return this.extractionMode === 'nonPriority' - ? { nonPriorityLogExtractionState: state } - : { logExtractionState: state }; + return { [LogsExtractionClient.CURSOR_FIELD[this.extractionMode]]: state } as Partial; } private async getLogExtractionConfigAndState( From dd7d48b11c403e84d259413d570b67ad959b1934 Mon Sep 17 00:00:00 2001 From: kibanamachine <42973632+kibanamachine@users.noreply.github.com> Date: Fri, 11 Sep 2026 13:27:56 +0000 Subject: [PATCH 8/8] Changes from node scripts/check --- .../server/domain/logs_extraction/logs_extraction_client.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts b/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts index 2f76356c9e169..153998db9917d 100644 --- a/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts +++ b/x-pack/platform/plugins/shared/entity_store/server/domain/logs_extraction/logs_extraction_client.ts @@ -142,7 +142,9 @@ export class LogsExtractionClient { }; private cursorPatch(state: EngineLogExtractionState): Partial { - return { [LogsExtractionClient.CURSOR_FIELD[this.extractionMode]]: state } as Partial; + return { + [LogsExtractionClient.CURSOR_FIELD[this.extractionMode]]: state, + } as Partial; } private async getLogExtractionConfigAndState(