From 5380b705dc779d7e8e0771c81bb133c276e293c2 Mon Sep 17 00:00:00 2001 From: woksin Date: Sat, 10 Oct 2026 21:41:44 +0200 Subject: [PATCH 1/3] Improve Code Health of EventSequence, ChronicleClient and Reducers Behaviour-preserving extract-method refactoring guided by CodeScene findings (#244). --- Source/ChronicleClient.ts | 219 +++++---- Source/eventSequences/EventSequence.ts | 453 ++++++++++-------- Source/eventSequences/PreparedBatchAppend.ts | 7 + Source/eventSequences/PreparedSingleAppend.ts | 7 + Source/eventSequences/WireBatchEvent.ts | 7 + Source/eventSequences/toContractsCausation.ts | 17 + Source/reducers/ReducerObservation.ts | 25 + Source/reducers/ReducerObservationSession.ts | 15 + Source/reducers/Reducers.ts | 351 +++++++------- Source/reducers/ReductionOutcome.ts | 75 +++ 10 files changed, 718 insertions(+), 458 deletions(-) create mode 100644 Source/eventSequences/PreparedBatchAppend.ts create mode 100644 Source/eventSequences/PreparedSingleAppend.ts create mode 100644 Source/eventSequences/WireBatchEvent.ts create mode 100644 Source/eventSequences/toContractsCausation.ts create mode 100644 Source/reducers/ReducerObservation.ts create mode 100644 Source/reducers/ReducerObservationSession.ts create mode 100644 Source/reducers/ReductionOutcome.ts diff --git a/Source/ChronicleClient.ts b/Source/ChronicleClient.ts index 2dee6bb5..4d1327b8 100644 --- a/Source/ChronicleClient.ts +++ b/Source/ChronicleClient.ts @@ -27,6 +27,25 @@ import { TypeDiscoverer } from './types/index.js'; import { takeUnregisteredModelBoundMappings } from './types/modelBoundPropertyMetadata.js'; import { reachableReadModelTypes } from './readModels/rootReadModelTypes.js'; +const connectionIndicators = [ + 'deadline exceeded', + 'unavailable', + 'connection', + 'connectivity', + 'channel', + 'socket', + 'econnrefused', + 'etimedout' +]; + +const errorCodeOf = (error: unknown): number | undefined => (error as { code?: number })?.code; + +const isTerminalServerError = (error: unknown): error is IncompatibleChronicleServer | RejectedChronicleCredentials => + error instanceof IncompatibleChronicleServer || error instanceof RejectedChronicleCredentials; + +/** gRPC status codes DEADLINE_EXCEEDED (4), INTERNAL (13) and UNAVAILABLE (14) are worth a reconnect. */ +const hasReconnectableStatusCode = (error: unknown): boolean => [4, 13, 14].includes(Number(errorCodeOf(error) ?? -1)); + /** * Implements {@link IChronicleClient} by managing a gRPC connection to the * Chronicle Kernel via {@link ChronicleConnection}. @@ -118,46 +137,13 @@ export class ChronicleClient implements IChronicleClient { /** @inheritdoc */ async getEventStore(name: string | EventStoreName, namespace?: string | EventStoreNamespaceName): Promise { const storeName = typeof name === 'string' ? new EventStoreName(name) : name; - const namespaceName = namespace === undefined - ? EventStoreNamespaceName.default - : typeof namespace === 'string' - ? new EventStoreNamespaceName(namespace) - : namespace; + const namespaceName = this.toNamespaceName(namespace); return observeOperation('getEventStore', async span => { setTelemetryAttribute(span, 'eventStore', storeName.value); setTelemetryAttribute(span, 'namespace', namespaceName.value); try { - const store = await this.withReconnect('get_event_store', async () => { - await this.ensureConnected(); - - const key = `${storeName.value}/${namespaceName.value}`; - const existing = this._stores.get(key); - if (existing) { - this._logger.verbose('Returning cached event store', { - eventStore: storeName.value, - namespace: namespaceName.value - }); - return existing; - } - - this._logger.debug('Ensuring event store exists in kernel', { - eventStore: storeName.value - }); - ensureCommandSuccess('ensure event store', await this._connection.eventStores.ensureEventStore({ Name: storeName.value })); - - const created = new EventStore(storeName, namespaceName, this._connection, this._lifecycle, this.options.defaultSinkTypeId, this.options.clientArtifactsProvider, this.options.reactorResultHandler, this.options.artifactActivator, this.options.readModelNamingPolicy, this.options.telemetry); - this._stores.set(key, created); - - try { - await this.registerArtifactsForStore(created, 'new-store'); - return created; - } catch (error) { - if (this._stores.get(key) === created) this._stores.delete(key); - created.disposeObservations(); - throw error; - } - }); + const store = await this.withReconnect('get_event_store', () => this.getOrCreateEventStore(storeName, namespaceName)); ChronicleConventionMetrics.eventStoreRetrievals.add(1, { [names.attributes.eventStore]: storeName.value, @@ -179,6 +165,42 @@ export class ChronicleClient implements IChronicleClient { }); } + private toNamespaceName(namespace: string | EventStoreNamespaceName | undefined): EventStoreNamespaceName { + if (namespace === undefined) return EventStoreNamespaceName.default; + return typeof namespace === 'string' ? new EventStoreNamespaceName(namespace) : namespace; + } + + private async getOrCreateEventStore(storeName: EventStoreName, namespaceName: EventStoreNamespaceName): Promise { + await this.ensureConnected(); + + const key = `${storeName.value}/${namespaceName.value}`; + const existing = this._stores.get(key); + if (existing) { + this._logger.verbose('Returning cached event store', { + eventStore: storeName.value, + namespace: namespaceName.value + }); + return existing; + } + + this._logger.debug('Ensuring event store exists in kernel', { + eventStore: storeName.value + }); + ensureCommandSuccess('ensure event store', await this._connection.eventStores.ensureEventStore({ Name: storeName.value })); + + const created = new EventStore(storeName, namespaceName, this._connection, this._lifecycle, this.options.defaultSinkTypeId, this.options.clientArtifactsProvider, this.options.reactorResultHandler, this.options.artifactActivator, this.options.readModelNamingPolicy, this.options.telemetry); + this._stores.set(key, created); + + try { + await this.registerArtifactsForStore(created, 'new-store'); + return created; + } catch (error) { + if (this._stores.get(key) === created) this._stores.delete(key); + created.disposeObservations(); + throw error; + } + } + /** @inheritdoc */ async getEventStores(): Promise { return observeOperation('getEventStores', async span => { @@ -300,46 +322,7 @@ export class ChronicleClient implements IChronicleClient { if (this._connectionFailure) throw this._connectionFailure; if (!this._reconnectOperation) { - this._reconnectOperation = runInBackgroundContext(async () => { - this._logger.warn('Reconnecting to Chronicle kernel', { - reason, - error - }); - - if (this._lifecycle.isConnected) { - await this._lifecycle.disconnected(disconnectError => { - this._logger.error('Disconnected lifecycle callback failed', { - error: disconnectError - }); - }); - } - - let attempt = 0; - while (!this._isDisposed) { - try { - await this._connection.resetChannel(); - await this._connection.connect(); - await this._connection.server.getVersionInfo({}, { signal: AbortSignal.timeout(10_000) }); - this._consecutiveCredentialRejections = 0; - this._logger.info('Reconnected to Chronicle kernel', { attempt: attempt + 1 }); - await this.startKernelKeepAlive(); - await this._lifecycle.connected(connectedError => { - this._logger.error('Connected lifecycle callback failed after reconnect', { - error: connectedError - }); - }); - return; - } catch (reconnectError) { - const terminal = this.terminalConnectionError(reconnectError); - if (terminal) { - this.failConnection(terminal); - throw terminal; - } - attempt++; - await this.backOff(attempt, 'Reconnect attempt failed, retrying', reconnectError); - } - } - }).finally(() => { + this._reconnectOperation = runInBackgroundContext(() => this.recoverConnection(reason, error)).finally(() => { this._reconnectOperation = undefined; }); } @@ -347,6 +330,51 @@ export class ChronicleClient implements IChronicleClient { await this._reconnectOperation; } + private async recoverConnection(reason: string, error: unknown): Promise { + this._logger.warn('Reconnecting to Chronicle kernel', { + reason, + error + }); + + if (this._lifecycle.isConnected) { + await this._lifecycle.disconnected(disconnectError => { + this._logger.error('Disconnected lifecycle callback failed', { + error: disconnectError + }); + }); + } + + await this.reconnectWithRetry(); + } + + private async reconnectWithRetry(): Promise { + let attempt = 0; + while (!this._isDisposed) { + try { + await this._connection.resetChannel(); + await this._connection.connect(); + await this._connection.server.getVersionInfo({}, { signal: AbortSignal.timeout(10_000) }); + this._consecutiveCredentialRejections = 0; + this._logger.info('Reconnected to Chronicle kernel', { attempt: attempt + 1 }); + await this.startKernelKeepAlive(); + await this._lifecycle.connected(connectedError => { + this._logger.error('Connected lifecycle callback failed after reconnect', { + error: connectedError + }); + }); + return; + } catch (reconnectError) { + const terminal = this.terminalConnectionError(reconnectError); + if (terminal) { + this.failConnection(terminal); + throw terminal; + } + attempt++; + await this.backOff(attempt, 'Reconnect attempt failed, retrying', reconnectError); + } + } + } + /** * Waits out the exponential backoff for a failed attempt and reports why. * The delay is jittered so a fleet of clients that lost the same kernel does @@ -392,7 +420,7 @@ export class ChronicleClient implements IChronicleClient { } private terminalConnectionError(error: unknown): Error | undefined { - if (error instanceof IncompatibleChronicleServer || error instanceof RejectedChronicleCredentials) return error; + if (isTerminalServerError(error)) return error; const rejection = (error as { code?: number })?.code === 16 && (this.options.connectionString.apiKey || this.hasOAuthCredentialRejection(error)); this._consecutiveCredentialRejections = rejection ? this._consecutiveCredentialRejections + 1 : 0; @@ -410,29 +438,20 @@ export class ChronicleClient implements IChronicleClient { } private shouldReconnect(error: unknown): boolean { - if (error instanceof IncompatibleChronicleServer || error instanceof RejectedChronicleCredentials) return false; - if ((error as { code?: number })?.code === 16 && - (this.options.connectionString.apiKey || !!(error as { tokenFailure?: Error })?.tokenFailure)) return true; - const code = Number((error as { code?: number })?.code ?? -1); - const details = String((error as { details?: string })?.details ?? ''); - const message = this.toErrorMessage(error); + if (isTerminalServerError(error)) return false; + return this.isRejectedAuthentication(error) || + hasReconnectableStatusCode(error) || + this.mentionsConnectionProblem(error); + } - if (code === 4 || code === 13 || code === 14) { - return true; - } + private isRejectedAuthentication(error: unknown): boolean { + if (errorCodeOf(error) !== 16) return false; + return !!this.options.connectionString.apiKey || !!(error as { tokenFailure?: Error })?.tokenFailure; + } - const connectionIndicators = [ - 'deadline exceeded', - 'unavailable', - 'connection', - 'connectivity', - 'channel', - 'socket', - 'econnrefused', - 'etimedout' - ]; - - const combined = `${details} ${message}`.toLowerCase(); + private mentionsConnectionProblem(error: unknown): boolean { + const details = String((error as { details?: string })?.details ?? ''); + const combined = `${details} ${this.toErrorMessage(error)}`.toLowerCase(); return connectionIndicators.some(indicator => combined.includes(indicator)); } @@ -452,10 +471,14 @@ export class ChronicleClient implements IChronicleClient { this._watchdogHandle.unref?.(); } + private get cannotRunHealthCheck(): boolean { + return this._isDisposed || !this._lifecycle.isConnected || this._healthCheckInFlight; + } + private async runHealthCheck(): Promise { // A black-holed connection makes the probe itself hang, so without the // in-flight guard every tick would stack another one that never returns. - if (this._isDisposed || !this._lifecycle.isConnected || this._healthCheckInFlight) { + if (this.cannotRunHealthCheck) { return; } diff --git a/Source/eventSequences/EventSequence.ts b/Source/eventSequences/EventSequence.ts index 664236f0..ba57d1f5 100644 --- a/Source/eventSequences/EventSequence.ts +++ b/Source/eventSequences/EventSequence.ts @@ -11,6 +11,10 @@ import type { import { Constructor } from '@cratis/fundamentals'; import { prepareSingleAppend } from './prepareSingleAppend.js'; import { prepareBatchAppend } from './prepareBatchAppend.js'; +import type { PreparedBatchAppend } from './PreparedBatchAppend.js'; +import type { PreparedSingleAppend } from './PreparedSingleAppend.js'; +import type { WireBatchEvent } from './WireBatchEvent.js'; +import { toContractsCausation } from './toContractsCausation.js'; import { createAppendNotification, mapAppendNotificationCausation } from './createAppendNotification.js'; import { getEventTypeFor } from '../events/eventTypeDecorator.js'; import type { AppendedEvent } from '../events/AppendedEvent.js'; @@ -46,7 +50,7 @@ import { ChronicleConventionMetrics } from '../Metrics.js'; import { identityProvider, Identity } from '../identity/index.js'; import { causationManager, CausationType } from '../auditing/index.js'; import { toContractsGuid } from '../connection/Guid.js'; -import { ensureCommandResponse, ensureCommandSuccess, ensureQuerySuccess } from '../connection/callResults.js'; +import { ensureCommandResponse, ensureCommandSuccess, ensureQuerySuccess, type CallResultLike } from '../connection/callResults.js'; import type { ConcurrencyScope } from './ConcurrencyScope.js'; import type { IEventSources } from '../eventSources/IEventSources.js'; import { deriveConcurrencyScope, planDerivedGuards, resolveBatchRouting, resolveSingleRouting } from './resolveEventSourceRouting.js'; @@ -77,13 +81,10 @@ export class EventSequence implements IEventSequence { /** @inheritdoc */ async append(eventSourceId: string, event: object, options?: AppendOptions): Promise { const routing = resolveSingleRouting(this._eventSources, options); - const { eventType, correlationId, content, tags, namedTags, causationChain, identity, subject } = prepareSingleAppend(event, options); + const prepared = prepareSingleAppend(event, options); + const { eventType, correlationId, tags, namedTags, causationChain } = prepared; - const sequenceMetricAttributes = { - [names.attributes.eventStore]: this._eventStoreName, - [names.attributes.namespace]: this._namespace, - [names.attributes.eventSequenceId]: this.id.value - }; + const sequenceMetricAttributes = this.sequenceMetricAttributes; const metricAttributes = { ...sequenceMetricAttributes, [names.attributes.eventTypeId]: eventType.id.value }; return observeOperation('append', async span => { @@ -94,40 +95,7 @@ export class EventSequence implements IEventSequence { if (options?.sourceType) span.setAttribute(names.attributes.eventSourceType, options.sourceType); const startTime = performance.now(); try { - const concurrencyScope = options?.concurrencyScope ?? (routing - ? await deriveConcurrencyScope(routing, eventSourceId, options?.streamId, this.tailReader) - : undefined); - const request = { - EventStore: this._eventStoreName, - Namespace: this._namespace, - EventSequenceId: this.id.value, - CorrelationId: toContractsGuid(correlationId), - EventSourceType: routing?.sourceType ?? options?.sourceType, - EventSourceId: eventSourceId, - EventStreamType: routing?.streamType ?? options?.streamType, - EventStreamId: options?.streamId, - EventType: { - Id: eventType.id.value, - Generation: eventType.generation.value, - Tombstone: eventType.tombstone - }, - Content: content, - Causation: causationChain.map(c => ({ - Occurred: { Value: c.occurred.toISOString() }, - Type: c.type.name, - Properties: { ...c.properties } - })), - CausedBy: toContractsCausedBy(identity), - ConcurrencyScope: this.toContractConcurrencyScope(concurrencyScope), - EventSource: routing?.eventSource, - Tags: tags, - Occurred: options?.occurred === undefined ? undefined : { Value: options.occurred.toISOString() }, - Subject: subject ?? eventSourceId - }; - // Named tags travel on the dedicated call so a kernel without support rejects them instead of dropping them. - const response = namedTags.length === 0 - ? await this._connection.eventSequences.append(request) - : await this._connection.eventSequences.appendWithNamedTags({ ...request, NamedTags: toContractsNamedTags(namedTags) }); + const response = await this.sendAppend(eventSourceId, prepared, routing, options); const appendResponse = ensureCommandResponse('append event', response); const durationInSeconds = (performance.now() - startTime) / 1000; @@ -140,14 +108,7 @@ export class EventSequence implements IEventSequence { setSequenceNumber(span, result.sequenceNumber.value); span.setStatus({ code: SpanStatusCode.OK }); - ChronicleConventionMetrics.eventsAppended.add(1, metricAttributes); - ChronicleConventionMetrics.appendDuration.record(durationInSeconds, metricAttributes); - if (result.constraintViolations.length > 0) { - ChronicleConventionMetrics.constraintViolations.add(result.constraintViolations.length, sequenceMetricAttributes); - } - if (result.errors.length > 0) { - ChronicleConventionMetrics.appendErrors.add(result.errors.length, sequenceMetricAttributes); - } + this.recordAppendMetrics(result, durationInSeconds, metricAttributes, sequenceMetricAttributes); if (this.appendOperations.hasSubscribers) { this.appendOperations.publish([createAppendNotification(eventSourceId, event, result, @@ -165,6 +126,83 @@ export class EventSequence implements IEventSequence { }, correlationId); } + private get sequenceMetricAttributes(): Record { + return { + [names.attributes.eventStore]: this._eventStoreName, + [names.attributes.namespace]: this._namespace, + [names.attributes.eventSequenceId]: this.id.value + }; + } + + private async sendAppend( + eventSourceId: string, + prepared: PreparedSingleAppend, + routing: ReturnType, + options?: AppendOptions + ) { + const concurrencyScope = options?.concurrencyScope ?? (routing + ? await deriveConcurrencyScope(routing, eventSourceId, options?.streamId, this.tailReader) + : undefined); + const request = this.createAppendRequest(eventSourceId, prepared, concurrencyScope, this.toAppendRouteFields(routing, options)); + // Named tags travel on the dedicated call so a kernel without support rejects them instead of dropping them. + return prepared.namedTags.length === 0 + ? await this._connection.eventSequences.append(request) + : await this._connection.eventSequences.appendWithNamedTags({ ...request, NamedTags: toContractsNamedTags(prepared.namedTags) }); + } + + private toAppendRouteFields(routing: ReturnType, options?: AppendOptions) { + return { + EventSourceType: routing?.sourceType ?? options?.sourceType, + EventStreamType: routing?.streamType ?? options?.streamType, + EventStreamId: options?.streamId, + EventSource: routing?.eventSource, + Occurred: toContractsOccurred(options?.occurred) + }; + } + + private createAppendRequest( + eventSourceId: string, + { eventType, correlationId, content, tags, causationChain, identity, subject }: PreparedSingleAppend, + concurrencyScope: ConcurrencyScope | undefined, + routeFields: ReturnType + ) { + return { + EventStore: this._eventStoreName, + Namespace: this._namespace, + EventSequenceId: this.id.value, + CorrelationId: toContractsGuid(correlationId), + EventSourceId: eventSourceId, + EventType: { + Id: eventType.id.value, + Generation: eventType.generation.value, + Tombstone: eventType.tombstone + }, + Content: content, + Causation: toContractsCausation(causationChain), + CausedBy: toContractsCausedBy(identity), + ConcurrencyScope: this.toContractConcurrencyScope(concurrencyScope), + Tags: tags, + Subject: subject ?? eventSourceId, + ...routeFields + }; + } + + private recordAppendMetrics( + result: AppendResult, + durationInSeconds: number, + metricAttributes: Record, + sequenceMetricAttributes: Record + ): void { + ChronicleConventionMetrics.eventsAppended.add(1, metricAttributes); + ChronicleConventionMetrics.appendDuration.record(durationInSeconds, metricAttributes); + if (result.constraintViolations.length > 0) { + ChronicleConventionMetrics.constraintViolations.add(result.constraintViolations.length, sequenceMetricAttributes); + } + if (result.errors.length > 0) { + ChronicleConventionMetrics.appendErrors.add(result.errors.length, sequenceMetricAttributes); + } + } + /** @inheritdoc */ async appendMany(eventSourceId: string, events: object[], options?: AppendOptions): Promise; async appendMany(events: EventForEventSourceId[], options?: AppendOptions): Promise; @@ -173,21 +211,17 @@ export class EventSequence implements IEventSequence { eventsOrOptions?: object[] | AppendOptions, options?: AppendOptions ): Promise { - const { eventsForEventSourceIds, appendOptions, correlationId, batchCausationChain, identity, concurrencyScopes, eventsToAppend, namedTags } = - prepareBatchAppend(eventSourceIdOrEvents, eventsOrOptions, options); + const prepared = prepareBatchAppend(eventSourceIdOrEvents, eventsOrOptions, options); + const { eventsForEventSourceIds, appendOptions, correlationId, concurrencyScopes, eventsToAppend, namedTags } = prepared; - const wireEvents: Array<(typeof eventsToAppend)[number] & { EventSource?: string }> = eventsToAppend; + const wireEvents: WireBatchEvent[] = eventsToAppend; const routings = eventsForEventSourceIds.map(event => resolveBatchRouting(this._eventSources, event, appendOptions)); ensureNamedTagBatchIsSupported(namedTags.some(tagsForEvent => tagsForEvent.length > 0), routings.some(routing => routing !== undefined)); const explicitIds = new Set([...concurrencyScopes].filter(([, scope]) => scope !== undefined).map(([id]) => id)); const derivedGuards = planDerivedGuards(routings, eventsForEventSourceIds, appendOptions?.streamId, explicitIds); const distinctEventSourceIds = [...new Set(eventsForEventSourceIds.map(_ => _.eventSourceId))]; - const batchMetricAttributes = { - [names.attributes.eventStore]: this._eventStoreName, - [names.attributes.namespace]: this._namespace, - [names.attributes.eventSequenceId]: this.id.value - }; + const batchMetricAttributes = this.sequenceMetricAttributes; return observeOperation('appendMany', async span => { this.setSequenceAttributes(span); @@ -197,91 +231,19 @@ export class EventSequence implements IEventSequence { setTelemetryAttribute(span, 'eventCount', eventsForEventSourceIds.length); const startTime = performance.now(); try { - routings.forEach((routing, index) => { - if (!routing) return; - wireEvents[index] = { - ...wireEvents[index], - EventSourceType: routing.sourceType, - EventStreamType: routing.streamType ?? wireEvents[index].EventStreamType, - EventSource: routing.eventSource - }; - }); + this.applyBatchRouting(wireEvents, routings); for (const [id, { routing, streamId }] of derivedGuards) { concurrencyScopes.set(id, await deriveConcurrencyScope(routing, id, streamId, this.tailReader)); } - const request = { - EventStore: this._eventStoreName, - Namespace: this._namespace, - EventSequenceId: this.id.value, - CorrelationId: toContractsGuid(correlationId), - Events: wireEvents, - Causation: batchCausationChain.map(c => ({ - Occurred: { Value: c.occurred.toISOString() }, - Type: c.type.name, - Properties: { ...c.properties } - })), - CausedBy: toContractsCausedBy(identity), - ConcurrencyScopes: [...concurrencyScopes].map(([eventSourceId, scope]) => ({ - EventSourceId: eventSourceId, - Scope: this.toContractConcurrencyScope(scope) - })) - }; - const response = namedTags.every(tagsForEvent => tagsForEvent.length === 0) - ? await this._connection.eventSequences.appendManyForEventSources(request) - : await this._connection.eventSequences.appendManyForEventSourcesWithNamedTags({ - ...request, - Events: wireEvents.map((wireEvent, index) => ({ ...wireEvent, NamedTags: toContractsNamedTags(namedTags[index]) })) - }); + const response = await this.sendAppendMany(prepared, wireEvents); const appendManyResponse = ensureCommandResponse('append many events', response); const durationInSeconds = (performance.now() - startTime) / 1000; - // Mirrors the C# client: every per-event AppendResult in a batch carries all - // constraint violations and the first concurrency violation of the whole batch — - // the wire response doesn't correlate either back to a specific event index. - const firstConcurrencyViolation: ContractsAppendManyResponse['ConcurrencyViolations'][number] | undefined = - (appendManyResponse.ConcurrencyViolations ?? [])[0]; - const sequenceNumbers: ContractsAppendManyResponse['SequenceNumbers'] = appendManyResponse.SequenceNumbers ?? []; - const constraintViolations: ContractsAppendManyResponse['ConstraintViolations'] = appendManyResponse.ConstraintViolations ?? []; - const errors: ContractsAppendManyResponse['Errors'] = appendManyResponse.Errors ?? []; - const batchWasRejected = sequenceNumbers.length === 0 && - (constraintViolations.length > 0 || errors.length > 0 || firstConcurrencyViolation !== undefined); - if (sequenceNumbers.length === 0 && eventsForEventSourceIds.length > 0 && !batchWasRejected) { - throw new Error('Append many events returned no sequence numbers or rejection details.'); - } - const result = batchWasRejected - ? eventsForEventSourceIds.map(() => this.mapAppendResponse(0n, constraintViolations, errors, firstConcurrencyViolation)) - : sequenceNumbers.map((sequenceNumber, index) => - this.mapAppendResponse( - sequenceNumber, - constraintViolations, - errors.filter((_, errorIndex) => errorIndex === index), - firstConcurrencyViolation - ) - ); + const result = this.toAppendManyResults(appendManyResponse, eventsForEventSourceIds.length); span.setStatus({ code: SpanStatusCode.OK }); - ChronicleConventionMetrics.batchAppendsPerformed.add(1, batchMetricAttributes); - ChronicleConventionMetrics.eventsAppended.add(eventsForEventSourceIds.length, batchMetricAttributes); - ChronicleConventionMetrics.appendManyDuration.record(durationInSeconds, batchMetricAttributes); - - const totalViolations = result.reduce((sum: number, appendResult: AppendResult) => sum + appendResult.constraintViolations.length, 0); - if (totalViolations > 0) { - ChronicleConventionMetrics.constraintViolations.add(totalViolations, batchMetricAttributes); - } - const totalErrors = result.reduce((sum: number, appendResult: AppendResult) => sum + appendResult.errors.length, 0); - if (totalErrors > 0) { - ChronicleConventionMetrics.appendErrors.add(totalErrors, batchMetricAttributes); - } - - if (this.appendOperations.hasSubscribers && result.length > 0) { - const occurredAt = new Date(); - const causationEntries = mapAppendNotificationCausation(batchCausationChain); - this.appendOperations.publish(result.map((appendResult: AppendResult, index: number) => { - const { eventSourceId, event } = eventsForEventSourceIds[index]; - return createAppendNotification(eventSourceId, event, appendResult, correlationId.toString(), - causationEntries, eventsToAppend[index].Tags, occurredAt, namedTags[index]); - })); - } + this.recordAppendManyMetrics(result, eventsForEventSourceIds.length, durationInSeconds, batchMetricAttributes); + this.publishAppendMany(prepared, result); return result; } catch (error) { @@ -294,6 +256,106 @@ export class EventSequence implements IEventSequence { }, correlationId); } + private applyBatchRouting(wireEvents: WireBatchEvent[], routings: ReturnType[]): void { + routings.forEach((routing, index) => { + if (!routing) return; + wireEvents[index] = { + ...wireEvents[index], + EventSourceType: routing.sourceType, + EventStreamType: routing.streamType ?? wireEvents[index].EventStreamType, + EventSource: routing.eventSource + }; + }); + } + + private async sendAppendMany( + { correlationId, batchCausationChain, identity, concurrencyScopes, namedTags }: PreparedBatchAppend, + wireEvents: WireBatchEvent[] + ) { + const request = { + EventStore: this._eventStoreName, + Namespace: this._namespace, + EventSequenceId: this.id.value, + CorrelationId: toContractsGuid(correlationId), + Events: wireEvents, + Causation: toContractsCausation(batchCausationChain), + CausedBy: toContractsCausedBy(identity), + ConcurrencyScopes: [...concurrencyScopes].map(([eventSourceId, scope]) => ({ + EventSourceId: eventSourceId, + Scope: this.toContractConcurrencyScope(scope) + })) + }; + return namedTags.every(tagsForEvent => tagsForEvent.length === 0) + ? await this._connection.eventSequences.appendManyForEventSources(request) + : await this._connection.eventSequences.appendManyForEventSourcesWithNamedTags({ + ...request, + Events: wireEvents.map((wireEvent, index) => ({ ...wireEvent, NamedTags: toContractsNamedTags(namedTags[index]) })) + }); + } + + private toAppendManyResults(appendManyResponse: ContractsAppendManyResponse, eventCount: number): AppendResult[] { + // Mirrors the C# client: every per-event AppendResult in a batch carries all + // constraint violations and the first concurrency violation of the whole batch — + // the wire response doesn't correlate either back to a specific event index. + const firstConcurrencyViolation: ContractsAppendManyResponse['ConcurrencyViolations'][number] | undefined = + (appendManyResponse.ConcurrencyViolations ?? [])[0]; + const sequenceNumbers: ContractsAppendManyResponse['SequenceNumbers'] = appendManyResponse.SequenceNumbers ?? []; + const constraintViolations: ContractsAppendManyResponse['ConstraintViolations'] = appendManyResponse.ConstraintViolations ?? []; + const errors: ContractsAppendManyResponse['Errors'] = appendManyResponse.Errors ?? []; + if (sequenceNumbers.length === 0) { + const hasRejectionDetails = constraintViolations.length > 0 || errors.length > 0 || firstConcurrencyViolation !== undefined; + if (hasRejectionDetails) { + return Array.from({ length: eventCount }, () => this.mapAppendResponse(0n, constraintViolations, errors, firstConcurrencyViolation)); + } + if (eventCount > 0) { + throw new Error('Append many events returned no sequence numbers or rejection details.'); + } + } + return sequenceNumbers.map((sequenceNumber, index) => + this.mapAppendResponse( + sequenceNumber, + constraintViolations, + errors.filter((_, errorIndex) => errorIndex === index), + firstConcurrencyViolation + ) + ); + } + + private recordAppendManyMetrics( + result: AppendResult[], + eventCount: number, + durationInSeconds: number, + batchMetricAttributes: Record + ): void { + ChronicleConventionMetrics.batchAppendsPerformed.add(1, batchMetricAttributes); + ChronicleConventionMetrics.eventsAppended.add(eventCount, batchMetricAttributes); + ChronicleConventionMetrics.appendManyDuration.record(durationInSeconds, batchMetricAttributes); + + const totalViolations = result.reduce((sum: number, appendResult: AppendResult) => sum + appendResult.constraintViolations.length, 0); + if (totalViolations > 0) { + ChronicleConventionMetrics.constraintViolations.add(totalViolations, batchMetricAttributes); + } + const totalErrors = result.reduce((sum: number, appendResult: AppendResult) => sum + appendResult.errors.length, 0); + if (totalErrors > 0) { + ChronicleConventionMetrics.appendErrors.add(totalErrors, batchMetricAttributes); + } + } + + private publishAppendMany( + { eventsForEventSourceIds, correlationId, batchCausationChain, eventsToAppend, namedTags }: PreparedBatchAppend, + result: AppendResult[] + ): void { + if (!this.appendOperations.hasSubscribers || result.length === 0) return; + + const occurredAt = new Date(); + const causationEntries = mapAppendNotificationCausation(batchCausationChain); + this.appendOperations.publish(result.map((appendResult: AppendResult, index: number) => { + const { eventSourceId, event } = eventsForEventSourceIds[index]; + return createAppendNotification(eventSourceId, event, appendResult, correlationId.toString(), + causationEntries, eventsToAppend[index].Tags, occurredAt, namedTags[index]); + })); + } + /** @inheritdoc */ async getNextSequenceNumber(): Promise { const tail = await this.getTailSequenceNumber(); @@ -447,61 +509,53 @@ export class EventSequence implements IEventSequence { /** @inheritdoc */ async redact(sequenceNumber: EventSequenceNumber, reason: string): Promise { causationManager.add(CausationType.redact, { sequenceNumber: sequenceNumber.value.toString() }); - const causationChain = causationManager.getCurrentChain(); - const identity = identityProvider.getCurrent(); - return observeOperation('redact', async span => { - this.setSequenceAttributes(span); - setSequenceNumber(span, sequenceNumber.value); - try { - ensureCommandSuccess('redact event', await this._connection.eventSequences.redact({ - EventStore: this._eventStoreName, - Namespace: this._namespace, - EventSequenceId: this.id.value, - SequenceNumber: sequenceNumber.value, - Reason: reason, - Causation: causationChain.map(c => ({ - Occurred: { Value: c.occurred.toISOString() }, - Type: c.type.name, - Properties: { ...c.properties } - })), - CausedBy: toContractsCausedBy(identity) - })); - span.setStatus({ code: SpanStatusCode.OK }); - } catch (error) { - recordSafeException(span, error); - throw error; - } finally { - span.end(); - } - }); + return this.sendRedaction('redact', 'redact event', + span => setSequenceNumber(span, sequenceNumber.value), + causedBy => this._connection.eventSequences.redact({ + EventStore: this._eventStoreName, + Namespace: this._namespace, + EventSequenceId: this.id.value, + SequenceNumber: sequenceNumber.value, + Reason: reason, + ...causedBy + })); } /** @inheritdoc */ async redactForEventSource(eventSourceId: string, reason: string, eventTypes?: Constructor[]): Promise { causationManager.add(CausationType.redactForEventSource, { eventSourceId }); - const causationChain = causationManager.getCurrentChain(); - const identity = identityProvider.getCurrent(); const wireEventTypeIds = (eventTypes ?? []).map(constructor => getEventTypeFor(constructor as unknown as Function).id.value); - return observeOperation('redactForEventSource', async span => { + return this.sendRedaction('redactForEventSource', 'redact event source', + span => setEventSourceId(span, eventSourceId, this._telemetry), + causedBy => this._connection.eventSequences.redactForEventSource({ + EventStore: this._eventStoreName, + Namespace: this._namespace, + EventSequenceId: this.id.value, + EventSourceId: eventSourceId, + Reason: reason, + EventTypes: wireEventTypeIds, + ...causedBy + })); + } + + private async sendRedaction( + operationName: 'redact' | 'redactForEventSource', + description: string, + describeSpan: (span: Span) => void, + send: (causedBy: { Causation: ReturnType; CausedBy: object }) => Promise + ): Promise { + const causedBy = { + Causation: toContractsCausation(causationManager.getCurrentChain()), + CausedBy: toContractsCausedBy(identityProvider.getCurrent()) + }; + + return observeOperation(operationName, async span => { this.setSequenceAttributes(span); - setEventSourceId(span, eventSourceId, this._telemetry); + describeSpan(span); try { - ensureCommandSuccess('redact event source', await this._connection.eventSequences.redactForEventSource({ - EventStore: this._eventStoreName, - Namespace: this._namespace, - EventSequenceId: this.id.value, - EventSourceId: eventSourceId, - Reason: reason, - EventTypes: wireEventTypeIds, - Causation: causationChain.map(c => ({ - Occurred: { Value: c.occurred.toISOString() }, - Type: c.type.name, - Properties: { ...c.properties } - })), - CausedBy: toContractsCausedBy(identity) - })); + ensureCommandSuccess(description, await send(causedBy)); span.setStatus({ code: SpanStatusCode.OK }); } catch (error) { recordSafeException(span, error); @@ -607,8 +661,7 @@ export class EventSequence implements IEventSequence { return { isSuccess: true, failedPartitions: [] }; } - const timeoutMs = typeof options === 'number' ? options : options.timeoutMs ?? DEFAULT_WAIT_FOR_COMPLETION_TIMEOUT_MS; - const callerSignal = typeof options === 'number' ? undefined : options.signal; + const { timeoutMs, callerSignal } = this.resolveWaitOptions(options); if (callerSignal?.aborted) throw callerSignal.reason; const controller = new AbortController(); @@ -635,28 +688,52 @@ export class EventSequence implements IEventSequence { } } + private resolveWaitOptions(options: number | WaitForCompletionOptions): { timeoutMs: number; callerSignal: AbortSignal | undefined } { + if (typeof options === 'number') return { timeoutMs: options, callerSignal: undefined }; + return { timeoutMs: options.timeoutMs ?? DEFAULT_WAIT_FOR_COMPLETION_TIMEOUT_MS, callerSignal: options.signal }; + } + private readonly tailReader = (source?: string, sourceType?: string, streamType?: string, streamId?: string) => this.getTailSequenceNumber(source, sourceType, streamType, streamId); private toContractConcurrencyScope(scope?: ConcurrencyScope) { + const expectsNoMatchingEvent = scope?.sequenceNumber === EventSequenceNumber.beforeFirst.value; return { - SequenceNumber: scope?.sequenceNumber === EventSequenceNumber.beforeFirst.value + SequenceNumber: expectsNoMatchingEvent ? EventSequenceNumber.unset.value : scope?.sequenceNumber ?? EventSequenceNumber.unset.value, - ExpectsNoMatchingEvent: scope?.sequenceNumber === EventSequenceNumber.beforeFirst.value, + ExpectsNoMatchingEvent: expectsNoMatchingEvent, EventSourceId: scope?.eventSourceId ?? false, EventStreamType: scope?.eventStreamType ?? '', EventStreamId: scope?.eventStreamId ?? '', EventSourceType: scope?.eventSourceType ?? '', - EventTypes: (scope?.eventTypes ?? []).map(eventType => ({ - Id: eventType.id.value, - Generation: eventType.generation.value, - Tombstone: eventType.tombstone - })) + EventTypes: toContractsEventTypes(scope) }; } } +/** + * Converts an optional occurrence time into the contracts timestamp shape. + * @param occurred - The occurrence time, if any. + * @returns The contracts timestamp, or undefined when no time was given. + */ +function toContractsOccurred(occurred: Date | undefined): { Value: string } | undefined { + return occurred === undefined ? undefined : { Value: occurred.toISOString() }; +} + +/** + * Converts the event types of a concurrency scope into the contracts shape. + * @param scope - The concurrency scope, if any. + * @returns The contracts event types. + */ +function toContractsEventTypes(scope: ConcurrencyScope | undefined) { + return (scope?.eventTypes ?? []).map(eventType => ({ + Id: eventType.id.value, + Generation: eventType.generation.value, + Tombstone: eventType.tombstone + })); +} + /** * Converts structured named tags into the sequence contract shape. * @param namedTags - The validated named tags. diff --git a/Source/eventSequences/PreparedBatchAppend.ts b/Source/eventSequences/PreparedBatchAppend.ts new file mode 100644 index 00000000..a5f31c9d --- /dev/null +++ b/Source/eventSequences/PreparedBatchAppend.ts @@ -0,0 +1,7 @@ +// Copyright (c) Cratis. All rights reserved. +// Licensed under the MIT license. See LICENSE file in the project root for full license information. + +import type { prepareBatchAppend } from './prepareBatchAppend.js'; + +/** The serialized form of a batch append, ready to be sent to the kernel. */ +export type PreparedBatchAppend = ReturnType; diff --git a/Source/eventSequences/PreparedSingleAppend.ts b/Source/eventSequences/PreparedSingleAppend.ts new file mode 100644 index 00000000..44eb60da --- /dev/null +++ b/Source/eventSequences/PreparedSingleAppend.ts @@ -0,0 +1,7 @@ +// Copyright (c) Cratis. All rights reserved. +// Licensed under the MIT license. See LICENSE file in the project root for full license information. + +import type { prepareSingleAppend } from './prepareSingleAppend.js'; + +/** The serialized form of a single append, ready to be sent to the kernel. */ +export type PreparedSingleAppend = ReturnType; diff --git a/Source/eventSequences/WireBatchEvent.ts b/Source/eventSequences/WireBatchEvent.ts new file mode 100644 index 00000000..734e0180 --- /dev/null +++ b/Source/eventSequences/WireBatchEvent.ts @@ -0,0 +1,7 @@ +// Copyright (c) Cratis. All rights reserved. +// Licensed under the MIT license. See LICENSE file in the project root for full license information. + +import type { PreparedBatchAppend } from './PreparedBatchAppend.js'; + +/** A wire event of a batch append, optionally carrying the definition event source it is routed to. */ +export type WireBatchEvent = PreparedBatchAppend['eventsToAppend'][number] & { EventSource?: string }; diff --git a/Source/eventSequences/toContractsCausation.ts b/Source/eventSequences/toContractsCausation.ts new file mode 100644 index 00000000..775ab1a6 --- /dev/null +++ b/Source/eventSequences/toContractsCausation.ts @@ -0,0 +1,17 @@ +// Copyright (c) Cratis. All rights reserved. +// Licensed under the MIT license. See LICENSE file in the project root for full license information. + +import type { causationManager } from '../auditing/index.js'; + +/** + * Converts a causation chain into the Causation shape used by Chronicle contracts. + * @param causationChain - The causation chain to convert. + * @returns The contracts causation entries. + */ +export function toContractsCausation(causationChain: ReturnType) { + return causationChain.map(c => ({ + Occurred: { Value: c.occurred.toISOString() }, + Type: c.type.name, + Properties: { ...c.properties } + })); +} diff --git a/Source/reducers/ReducerObservation.ts b/Source/reducers/ReducerObservation.ts new file mode 100644 index 00000000..27cbf397 --- /dev/null +++ b/Source/reducers/ReducerObservation.ts @@ -0,0 +1,25 @@ +// Copyright (c) Cratis. All rights reserved. +// Licensed under the MIT license. See LICENSE file in the project root for full license information. + +import { Constructor } from '@cratis/fundamentals'; +import { ReducerEventDispatcher } from './ReducerEventDispatcher.js'; + +/** + * Identifies a single reducer observation: the reducer, where it observes and how its events are dispatched. + */ +export interface ReducerObservation { + /** The reducer identifier. */ + readonly id: string; + + /** The reducer type. */ + readonly reducerType: Constructor; + + /** The event sequence the reducer observes. */ + readonly eventSequenceId: string; + + /** The identifier of the read model the reducer produces. */ + readonly readModelName: string; + + /** The dispatcher routing events to the reducer's handlers. */ + readonly dispatcher: ReducerEventDispatcher; +} diff --git a/Source/reducers/ReducerObservationSession.ts b/Source/reducers/ReducerObservationSession.ts new file mode 100644 index 00000000..e8051f17 --- /dev/null +++ b/Source/reducers/ReducerObservationSession.ts @@ -0,0 +1,15 @@ +// Copyright (c) Cratis. All rights reserved. +// Licensed under the MIT license. See LICENSE file in the project root for full license information. + +import { ReducerObservation } from './ReducerObservation.js'; + +/** + * A {@link ReducerObservation} that is running against an open observation stream. + */ +export interface ReducerObservationSession extends ReducerObservation { + /** Aborted when the observation is stopped. */ + readonly signal: AbortSignal; + + /** The long-lived reducer instance, or undefined when instances are activated per delivery. */ + readonly reducerInstance: Record | undefined; +} diff --git a/Source/reducers/Reducers.ts b/Source/reducers/Reducers.ts index 4787163f..4c29d41b 100644 --- a/Source/reducers/Reducers.ts +++ b/Source/reducers/Reducers.ts @@ -5,7 +5,7 @@ import 'reflect-metadata'; import { createLogger } from '../logging/createLogger.js'; import { runInBackgroundContext } from '../telemetry/runInBackgroundContext.js'; import { Constructor } from '@cratis/fundamentals'; -import { ObservationState, ReadModelObserverType, ReducerMessage, ReplayState } from '@cratis/chronicle.contracts'; +import { AppendedEvent, ReadModelObserverType, ReduceOperationMessage, ReducerMessage, ReplayState } from '@cratis/chronicle.contracts'; import { IClientArtifactsProvider } from '../artifacts/index.js'; import { ChronicleConnection } from '../connection/index.js'; import { ConnectionLifecycle } from '../connection/ConnectionLifecycle.js'; @@ -19,6 +19,9 @@ import { notifyReplayLifecycle } from '../observation/notifyReplayLifecycle.js'; import { IReducers } from './IReducers.js'; import { getReducerMetadata } from './reducer.js'; import { ReducerEventDispatcher } from './ReducerEventDispatcher.js'; +import { ReducerObservation } from './ReducerObservation.js'; +import { ReducerObservationSession } from './ReducerObservationSession.js'; +import { ReductionOutcome } from './ReductionOutcome.js'; import { getReadModelMetadata } from '../readModels/index.js'; import { getReadModelId } from '../readModels/readModel.js'; import { buildReadModelDefinition } from '../readModels/buildReadModelDefinition.js'; @@ -31,14 +34,10 @@ import type { ActivatedArtifact } from '../artifacts/ActivatedArtifact.js'; import { ArtifactKind } from '../artifacts/ArtifactKind.js'; import { ArtifactDelivery } from '../artifacts/ArtifactDelivery.js'; import { withActivatedArtifact, runActivated } from '../artifacts/withActivatedArtifact.js'; -import { ArtifactCompletionFailed } from '../artifacts/ArtifactCompletionFailed.js'; /** Expression used to partition reducer observations by event source ID. */ const EVENT_SOURCE_ID_KEY = '$eventSourceId'; -/** Sentinel sequence number sent back when no event was successfully processed. */ -const SEQUENCE_NUMBER_UNAVAILABLE = 4294967295n; - /** * A push-based async queue that implements {@link AsyncIterable} for use with nice-grpc * bidirectional streaming. Values pushed via {@link send} are yielded in order to @@ -259,24 +258,18 @@ export class Reducers implements IReducers { handlers: dispatcher.handlers.map(e => e.methodName) }); - void this.runObservation(id, reducerType, eventSequenceId, readModelName, dispatcher); + void this.runObservation({ id, reducerType, eventSequenceId, readModelName, dispatcher }); }); } - private async runObservation( - id: string, - reducerType: Constructor, - eventSequenceId: string, - readModelName: string, - dispatcher: ReducerEventDispatcher - ): Promise { + private async runObservation(observation: ReducerObservation): Promise { try { - await this.observeReducer(id, reducerType, eventSequenceId, readModelName, dispatcher); + await this.observeReducer(observation); } catch (error) { - this._logger.error('Reducer observation loop exited with error', { reducerId: id, error }); + this._logger.error('Reducer observation loop exited with error', { reducerId: observation.id, error }); } - this.scheduleReobserve(id, reducerType, dispatcher); + this.scheduleReobserve(observation.id, observation.reducerType, observation.dispatcher); } /** @@ -317,20 +310,43 @@ export class Reducers implements IReducers { return (reducerType as Function).name; } - private async observeReducer( - id: string, - reducerType: Constructor, - eventSequenceId: string, - readModelName: string, - dispatcher: ReducerEventDispatcher - ): Promise { + private async observeReducer(observation: ReducerObservation): Promise { + const { id, reducerType } = observation; const queue = new AsyncQueue(); const controller = new AbortController(); this._queues.set(id, queue); this._observations.set(id, controller); - const isActive = getReducerMetadata(reducerType)?.isActive ?? true; - queue.send({ + queue.send(this.createRegistrationMessage(observation)); + + try { + const reducerInstance = this._artifactActivator ? undefined : new (reducerType as new () => Record)(); + if (reducerInstance) observation.dispatcher.validateInstance(reducerInstance); + const session: ReducerObservationSession = { ...observation, signal: controller.signal, reducerInstance }; + + for await (const operation of this._connection.reducers.observe(queue, { signal: controller.signal })) { + queue.send(await this.reduce(session, operation)); + } + } catch (err) { + if (!this._queues.has(id)) { + this._logger.debug('Reducer observation stream closed cleanly', { reducerId: id }); + } else { + this._logger.error('Reducer observation stream ended unexpectedly', { reducerId: id, error: err }); + } + } finally { + // Only retire our own queue: a reconnect can already have replaced it, + // and deleting the new one would leave that observation untracked and + // leak a duplicate stream on every reconnect. + if (this._queues.get(id) === queue) { + this._queues.delete(id); + this._observations.delete(id); + } + queue.complete(); + } + } + + private createRegistrationMessage({ id, reducerType, eventSequenceId, readModelName, dispatcher }: ReducerObservation): ReducerMessage { + return { Content: { Value0: { ConnectionId: this._lifecycle.connectionId, @@ -345,7 +361,7 @@ export class Reducers implements IReducers { })), ReadModel: readModelName, Hash: '', - IsActive: isActive, + IsActive: getReducerMetadata(reducerType)?.isActive ?? true, Tags: getTagsFor(reducerType).map(t => t.value), Filters: { FilterTags: getFilterTagsFor(reducerType).map(t => t.value), @@ -356,164 +372,155 @@ export class Reducers implements IReducers { }, Value1: undefined } - }); + }; + } - try { - const reducerInstance = this._artifactActivator ? undefined : new (reducerType as new () => Record)(); - if (reducerInstance) dispatcher.validateInstance(reducerInstance); + private async reduce(session: ReducerObservationSession, operation: ReduceOperationMessage): Promise { + const outcome = new ReductionOutcome(); - for await (const operation of this._connection.reducers.observe(queue, { signal: controller.signal })) { - let lastSuccessfullyObservedEvent = SEQUENCE_NUMBER_UNAVAILABLE; - let state = ObservationState.Success; - const exceptionMessages: string[] = []; - let exceptionStackTrace = ''; - let readModelState: string | undefined; - - this._logger.debug('Received reduce operation', { - reducerId: id, - count: operation.Events.length, - hasInitialState: operation.InitialState !== '', - replayState: operation.ReplayState - }); + this._logger.debug('Received reduce operation', { + reducerId: session.id, + count: operation.Events.length, + hasInitialState: operation.InitialState !== '', + replayState: operation.ReplayState + }); - try { - if (this._artifactActivator && operation.ReplayState !== ReplayState.REPLAY_STATE_None) { - if (!this._eventStore) throw new Error('Reducer activation requires the owning event store.'); - await withActivatedArtifact(reducerType, { - kind: ArtifactKind.Reducer, artifactId: id, eventStore: this._eventStore, - readModels: this._eventStore.readModels, eventSequenceId, partition: operation.Partition, - signal: controller.signal, delivery: ArtifactDelivery.ReplayNotification, - replayState: operation.ReplayState - }, this._artifactActivator, artifact => { - dispatcher.validateInstance(artifact.instance); - return runActivated(artifact, () => - notifyReplayLifecycle(artifact.instance, operation.ReplayState, operation.Partition), - { delivery: ArtifactDelivery.ReplayNotification, replayState: operation.ReplayState }); - }, this._connection.logger); - } else if (reducerInstance) { - await notifyReplayLifecycle(reducerInstance, operation.ReplayState, operation.Partition); - } - } catch (err) { - this._logger.error('Error notifying reducer of replay lifecycle transition', { reducerId: id, error: err }); - if (err instanceof ArtifactCompletionFailed && err.processingError !== undefined) { - exceptionMessages.push(String(err.processingError)); - } - exceptionMessages.push(String(err)); - exceptionStackTrace = err instanceof Error ? (err.stack ?? '') : ''; - state = ObservationState.Failed; - } + await this.notifyReplayTransition(session, operation, outcome); - let currentState: unknown = operation.InitialState - ? JSON.parse(operation.InitialState) as unknown - : undefined; - - const processEvents = async (artifact: ActivatedArtifact>) => { - for (const event of state === ObservationState.Failed ? [] : operation.Events) { - try { - const eventTypeId = event.Context?.EventType?.Id; - if (!eventTypeId) { - this._logger.warn('Event missing event type context', { reducerId: id }); - continue; - } - - const entry = dispatcher.handlerFor(eventTypeId); - if (!entry) { - this._logger.debug('No handler registered for event type — skipping', { reducerId: id, eventTypeId }); - lastSuccessfullyObservedEvent = event.Context!.SequenceNumber; - continue; - } - - const content = JSON.parse(event.Content) as Record; - const context = toClientEventContext(event.Context!); - - this._logger.info('Invoking reducer handler', { - reducerId: id, - method: entry.methodName, - sequenceNumber: event.Context!.SequenceNumber.toString(), - eventTypeId, - hasState: currentState !== undefined - }); - - currentState = await runActivated(artifact, () => dispatcher.invoke(artifact.instance, entry, content, currentState, context), - { delivery: ArtifactDelivery.Events, eventContext: context, methodName: entry.methodName }); - lastSuccessfullyObservedEvent = event.Context!.SequenceNumber; - } catch (err) { - this._logger.error('Error handling event in reducer', { reducerId: id, error: err }); - exceptionMessages.push(String(err)); - exceptionStackTrace = err instanceof Error ? (err.stack ?? '') : ''; - state = ObservationState.Failed; - break; - } - } - }; - - const firstInvocableEvent = this._artifactActivator && state === ObservationState.Success - ? operation.Events.find(event => - !!event.Context?.EventType?.Id && !!dispatcher.handlerFor(event.Context.EventType.Id)) - : undefined; - if (firstInvocableEvent && this._artifactActivator) { - try { - if (!this._eventStore) throw new Error('Reducer activation requires the owning event store.'); - await withActivatedArtifact(reducerType as new () => Record, { - kind: ArtifactKind.Reducer, artifactId: id, eventStore: this._eventStore, - readModels: this._eventStore.readModels, eventSequenceId, partition: operation.Partition, - signal: controller.signal, delivery: ArtifactDelivery.Events, - eventContext: toClientEventContext(firstInvocableEvent.Context!) - }, this._artifactActivator, artifact => { - dispatcher.validateInstance(artifact.instance); - return processEvents(artifact); - }, this._connection.logger); - } catch (err) { - this._logger.error('Error activating reducer', { reducerId: id, error: err }); - exceptionMessages.push(String(err)); - const completionStack = err instanceof Error ? (err.stack ?? '') : ''; - exceptionStackTrace = err instanceof ArtifactCompletionFailed && exceptionStackTrace - ? `${exceptionStackTrace}\n${completionStack}` : completionStack; - state = ObservationState.Failed; - if (err instanceof ArtifactCompletionFailed) { - lastSuccessfullyObservedEvent = SEQUENCE_NUMBER_UNAVAILABLE; - } - } - } else { - await processEvents({ instance: reducerInstance ?? {} }); - } + outcome.currentState = operation.InitialState + ? JSON.parse(operation.InitialState) as unknown + : undefined; - if (state === ObservationState.Success && currentState !== undefined) { - readModelState = JSON.stringify(currentState); - } + await this.reduceEvents(session, operation, outcome); - queue.send({ - Content: { - Value0: undefined, - Value1: { - Partition: operation.Partition, - State: state, - LastSuccessfulObservation: lastSuccessfullyObservedEvent, - ExceptionMessages: exceptionMessages, - ExceptionStackTrace: exceptionStackTrace, - ReadModelState: readModelState ?? '' - } - } - }); + return outcome.toMessage(operation.Partition); + } + + private async notifyReplayTransition( + session: ReducerObservationSession, + operation: ReduceOperationMessage, + outcome: ReductionOutcome + ): Promise { + try { + if (this._artifactActivator && operation.ReplayState !== ReplayState.REPLAY_STATE_None) { + await this.notifyActivatedReplayTransition(session, operation, this._artifactActivator); + } else if (session.reducerInstance) { + await notifyReplayLifecycle(session.reducerInstance, operation.ReplayState, operation.Partition); } } catch (err) { - if (!this._queues.has(id)) { - this._logger.debug('Reducer observation stream closed cleanly', { reducerId: id }); - } else { - this._logger.error('Reducer observation stream ended unexpectedly', { reducerId: id, error: err }); - } - } finally { - // Only retire our own queue: a reconnect can already have replaced it, - // and deleting the new one would leave that observation untracked and - // leak a duplicate stream on every reconnect. - if (this._queues.get(id) === queue) { - this._queues.delete(id); - this._observations.delete(id); + this._logger.error('Error notifying reducer of replay lifecycle transition', { reducerId: session.id, error: err }); + outcome.recordReplayNotificationFailure(err); + } + } + + private async notifyActivatedReplayTransition( + { id, reducerType, eventSequenceId, dispatcher, signal }: ReducerObservationSession, + operation: ReduceOperationMessage, + activator: ClientArtifactsActivator + ): Promise { + const eventStore = this.requireEventStore(); + await withActivatedArtifact(reducerType, { + kind: ArtifactKind.Reducer, artifactId: id, eventStore, + readModels: eventStore.readModels, eventSequenceId, partition: operation.Partition, + signal, delivery: ArtifactDelivery.ReplayNotification, + replayState: operation.ReplayState + }, activator, artifact => { + dispatcher.validateInstance(artifact.instance); + return runActivated(artifact, () => + notifyReplayLifecycle(artifact.instance, operation.ReplayState, operation.Partition), + { delivery: ArtifactDelivery.ReplayNotification, replayState: operation.ReplayState }); + }, this._connection.logger); + } + + private requireEventStore(): IEventStore { + if (!this._eventStore) throw new Error('Reducer activation requires the owning event store.'); + return this._eventStore; + } + + private async reduceEvents( + session: ReducerObservationSession, + operation: ReduceOperationMessage, + outcome: ReductionOutcome + ): Promise { + const { id, reducerType, eventSequenceId, dispatcher, signal, reducerInstance } = session; + const firstInvocableEvent = this._artifactActivator && !outcome.failed + ? operation.Events.find(event => + !!event.Context?.EventType?.Id && !!dispatcher.handlerFor(event.Context.EventType.Id)) + : undefined; + if (!firstInvocableEvent || !this._artifactActivator) { + await this.processEvents(session, operation, outcome, { instance: reducerInstance ?? {} }); + return; + } + + try { + const eventStore = this.requireEventStore(); + await withActivatedArtifact(reducerType as new () => Record, { + kind: ArtifactKind.Reducer, artifactId: id, eventStore, + readModels: eventStore.readModels, eventSequenceId, partition: operation.Partition, + signal, delivery: ArtifactDelivery.Events, + eventContext: toClientEventContext(firstInvocableEvent.Context!) + }, this._artifactActivator, artifact => { + dispatcher.validateInstance(artifact.instance); + return this.processEvents(session, operation, outcome, artifact); + }, this._connection.logger); + } catch (err) { + this._logger.error('Error activating reducer', { reducerId: id, error: err }); + outcome.recordActivationFailure(err); + } + } + + private async processEvents( + session: ReducerObservationSession, + operation: ReduceOperationMessage, + outcome: ReductionOutcome, + artifact: ActivatedArtifact> + ): Promise { + for (const event of outcome.failed ? [] : operation.Events) { + try { + await this.processEvent(session, event, outcome, artifact); + } catch (err) { + this._logger.error('Error handling event in reducer', { reducerId: session.id, error: err }); + outcome.recordFailure(err); + break; } - queue.complete(); } } + private async processEvent( + { id, dispatcher }: ReducerObservationSession, + event: AppendedEvent, + outcome: ReductionOutcome, + artifact: ActivatedArtifact> + ): Promise { + const eventTypeId = event.Context?.EventType?.Id; + if (!eventTypeId) { + this._logger.warn('Event missing event type context', { reducerId: id }); + return; + } + + const entry = dispatcher.handlerFor(eventTypeId); + if (!entry) { + this._logger.debug('No handler registered for event type — skipping', { reducerId: id, eventTypeId }); + outcome.lastSuccessfullyObservedEvent = event.Context!.SequenceNumber; + return; + } + + const content = JSON.parse(event.Content) as Record; + const context = toClientEventContext(event.Context!); + + this._logger.info('Invoking reducer handler', { + reducerId: id, + method: entry.methodName, + sequenceNumber: event.Context!.SequenceNumber.toString(), + eventTypeId, + hasState: outcome.currentState !== undefined + }); + + outcome.currentState = await runActivated(artifact, () => dispatcher.invoke(artifact.instance, entry, content, outcome.currentState, context), + { delivery: ArtifactDelivery.Events, eventContext: context, methodName: entry.methodName }); + outcome.lastSuccessfullyObservedEvent = event.Context!.SequenceNumber; + } + private disconnectAll(): void { for (const controller of this._observations.values()) controller.abort(); for (const queue of this._queues.values()) queue.complete(); diff --git a/Source/reducers/ReductionOutcome.ts b/Source/reducers/ReductionOutcome.ts new file mode 100644 index 00000000..79171167 --- /dev/null +++ b/Source/reducers/ReductionOutcome.ts @@ -0,0 +1,75 @@ +// Copyright (c) Cratis. All rights reserved. +// Licensed under the MIT license. See LICENSE file in the project root for full license information. + +import { ObservationState, ReducerMessage } from '@cratis/chronicle.contracts'; +import { ArtifactCompletionFailed } from '../artifacts/ArtifactCompletionFailed.js'; + +/** Sentinel sequence number sent back when no event was successfully processed. */ +export const SEQUENCE_NUMBER_UNAVAILABLE = 4294967295n; + +const stackOf = (error: unknown): string => error instanceof Error ? (error.stack ?? '') : ''; + +/** + * Accumulates the result of reducing one reduce operation: the state, the failure details and the + * last event that was successfully reduced. + */ +export class ReductionOutcome { + state = ObservationState.Success; + lastSuccessfullyObservedEvent = SEQUENCE_NUMBER_UNAVAILABLE; + readonly exceptionMessages: string[] = []; + exceptionStackTrace = ''; + currentState: unknown; + + /** Gets whether the reduction has failed. */ + get failed(): boolean { + return this.state === ObservationState.Failed; + } + + /** Records a failure of the reduction. */ + recordFailure(error: unknown): void { + this.exceptionMessages.push(String(error)); + this.exceptionStackTrace = stackOf(error); + this.state = ObservationState.Failed; + } + + /** Records a failure to notify of a replay lifecycle transition. */ + recordReplayNotificationFailure(error: unknown): void { + if (error instanceof ArtifactCompletionFailed && error.processingError !== undefined) { + this.exceptionMessages.push(String(error.processingError)); + } + this.recordFailure(error); + } + + /** Records a failure to activate the reducer or to complete its activation. */ + recordActivationFailure(error: unknown): void { + this.exceptionMessages.push(String(error)); + const completionStack = stackOf(error); + this.exceptionStackTrace = error instanceof ArtifactCompletionFailed && this.exceptionStackTrace + ? `${this.exceptionStackTrace}\n${completionStack}` : completionStack; + this.state = ObservationState.Failed; + if (error instanceof ArtifactCompletionFailed) { + this.lastSuccessfullyObservedEvent = SEQUENCE_NUMBER_UNAVAILABLE; + } + } + + /** Creates the result message reported back to the kernel. */ + toMessage(partition: string): ReducerMessage { + const readModelState = this.state === ObservationState.Success && this.currentState !== undefined + ? JSON.stringify(this.currentState) + : undefined; + + return { + Content: { + Value0: undefined, + Value1: { + Partition: partition, + State: this.state, + LastSuccessfulObservation: this.lastSuccessfullyObservedEvent, + ExceptionMessages: this.exceptionMessages, + ExceptionStackTrace: this.exceptionStackTrace, + ReadModelState: readModelState ?? '' + } + } + }; + } +} From eb71636d96f6e074b5d1de7b45235c2afcce3da2 Mon Sep 17 00:00:00 2001 From: woksin Date: Sat, 10 Oct 2026 22:06:30 +0200 Subject: [PATCH 2/3] Preserve async ordering and redaction telemetry in refactored code Keep skipped reducer events synchronous and serialize redaction causation inside the observed operation. --- Source/eventSequences/EventSequence.ts | 10 ++++++---- Source/reducers/Reducers.ts | 21 +++++++++++++++------ 2 files changed, 21 insertions(+), 10 deletions(-) diff --git a/Source/eventSequences/EventSequence.ts b/Source/eventSequences/EventSequence.ts index ba57d1f5..0221af42 100644 --- a/Source/eventSequences/EventSequence.ts +++ b/Source/eventSequences/EventSequence.ts @@ -546,15 +546,17 @@ export class EventSequence implements IEventSequence { describeSpan: (span: Span) => void, send: (causedBy: { Causation: ReturnType; CausedBy: object }) => Promise ): Promise { - const causedBy = { - Causation: toContractsCausation(causationManager.getCurrentChain()), - CausedBy: toContractsCausedBy(identityProvider.getCurrent()) - }; + const causationChain = causationManager.getCurrentChain(); + const identity = identityProvider.getCurrent(); return observeOperation(operationName, async span => { this.setSequenceAttributes(span); describeSpan(span); try { + const causedBy = { + Causation: toContractsCausation(causationChain), + CausedBy: toContractsCausedBy(identity) + }; ensureCommandSuccess(description, await send(causedBy)); span.setStatus({ code: SpanStatusCode.OK }); } catch (error) { diff --git a/Source/reducers/Reducers.ts b/Source/reducers/Reducers.ts index 4c29d41b..084bc5c6 100644 --- a/Source/reducers/Reducers.ts +++ b/Source/reducers/Reducers.ts @@ -477,7 +477,9 @@ export class Reducers implements IReducers { ): Promise { for (const event of outcome.failed ? [] : operation.Events) { try { - await this.processEvent(session, event, outcome, artifact); + // Skipped events stay synchronous; only an actual handler invocation is awaited. + const pending = this.processEvent(session, event, outcome, artifact); + if (pending) await pending; } catch (err) { this._logger.error('Error handling event in reducer', { reducerId: session.id, error: err }); outcome.recordFailure(err); @@ -486,12 +488,12 @@ export class Reducers implements IReducers { } } - private async processEvent( + private processEvent( { id, dispatcher }: ReducerObservationSession, event: AppendedEvent, outcome: ReductionOutcome, artifact: ActivatedArtifact> - ): Promise { + ): Promise | undefined { const eventTypeId = event.Context?.EventType?.Id; if (!eventTypeId) { this._logger.warn('Event missing event type context', { reducerId: id }); @@ -516,9 +518,16 @@ export class Reducers implements IReducers { hasState: outcome.currentState !== undefined }); - outcome.currentState = await runActivated(artifact, () => dispatcher.invoke(artifact.instance, entry, content, outcome.currentState, context), - { delivery: ArtifactDelivery.Events, eventContext: context, methodName: entry.methodName }); - outcome.lastSuccessfullyObservedEvent = event.Context!.SequenceNumber; + return this.invokeHandler( + () => runActivated(artifact, () => dispatcher.invoke(artifact.instance, entry, content, outcome.currentState, context), + { delivery: ArtifactDelivery.Events, eventContext: context, methodName: entry.methodName }), + outcome, + event.Context!.SequenceNumber); + } + + private async invokeHandler(invoke: () => Promise, outcome: ReductionOutcome, sequenceNumber: bigint): Promise { + outcome.currentState = await invoke(); + outcome.lastSuccessfullyObservedEvent = sequenceNumber; } private disconnectAll(): void { From 2bee0696042e0990fccde03ce948207576d71d2c Mon Sep 17 00:00:00 2001 From: woksin Date: Sun, 11 Oct 2026 10:01:09 +0200 Subject: [PATCH 3/3] Keep reducer delivery steps in the observation loop to preserve async context propagation --- Source/reducers/Reducers.ts | 24 +++++++++++------------- 1 file changed, 11 insertions(+), 13 deletions(-) diff --git a/Source/reducers/Reducers.ts b/Source/reducers/Reducers.ts index 084bc5c6..c2f701a9 100644 --- a/Source/reducers/Reducers.ts +++ b/Source/reducers/Reducers.ts @@ -325,7 +325,15 @@ export class Reducers implements IReducers { const session: ReducerObservationSession = { ...observation, signal: controller.signal, reducerInstance }; for await (const operation of this._connection.reducers.observe(queue, { signal: controller.signal })) { - queue.send(await this.reduce(session, operation)); + // Delivery steps are awaited here, not inside a wrapper async function, so that + // async-context updates made by handlers reach this loop as they did originally. + const outcome = this.beginReduction(session, operation); + await this.notifyReplayTransition(session, operation, outcome); + outcome.currentState = operation.InitialState + ? JSON.parse(operation.InitialState) as unknown + : undefined; + await this.reduceEvents(session, operation, outcome); + queue.send(outcome.toMessage(operation.Partition)); } } catch (err) { if (!this._queues.has(id)) { @@ -375,9 +383,7 @@ export class Reducers implements IReducers { }; } - private async reduce(session: ReducerObservationSession, operation: ReduceOperationMessage): Promise { - const outcome = new ReductionOutcome(); - + private beginReduction(session: ReducerObservationSession, operation: ReduceOperationMessage): ReductionOutcome { this._logger.debug('Received reduce operation', { reducerId: session.id, count: operation.Events.length, @@ -385,15 +391,7 @@ export class Reducers implements IReducers { replayState: operation.ReplayState }); - await this.notifyReplayTransition(session, operation, outcome); - - outcome.currentState = operation.InitialState - ? JSON.parse(operation.InitialState) as unknown - : undefined; - - await this.reduceEvents(session, operation, outcome); - - return outcome.toMessage(operation.Partition); + return new ReductionOutcome(); } private async notifyReplayTransition(