diff --git a/src/data-pump/data-pump.ts b/src/data-pump/data-pump.ts index 53dcef9..98f6748 100644 --- a/src/data-pump/data-pump.ts +++ b/src/data-pump/data-pump.ts @@ -15,6 +15,8 @@ import type { FlowcoreLogger, } from "./types.ts" +const textEncoder = new TextEncoder() + interface FlowcoreDataPumpNotifierNatsOptions { type: "nats" servers: string[] @@ -78,9 +80,17 @@ interface FlowcoreDataPumpBufferItem { event: FlowcoreEvent status: "open" | "reserved" deliveryCount: number + payloadSizeBytes: number + eventSizeBytes?: number deliveryId?: string } +interface FlowcoreDataPumpBufferStats { + eventCount: number + eventReservedCount: number + eventSizeBytes: number +} + export class FlowcoreDataPump { private nextCursor?: string private running = false @@ -99,6 +109,10 @@ export class FlowcoreDataPump { private processLoopRestartAttempts = 0 private mainLoopRestartAttempts = 0 private readonly replayObserver: ReplayStageObserver + private readonly bufferStats = new Map() + private bufferReservedCount = 0 + private bufferSizeBytes = 0 + private gaugePublicationScheduled = false private constructor( public readonly dataSource: FlowcoreDataSource, @@ -112,6 +126,9 @@ export class FlowcoreDataPump { data_core: this.dataSource.dataCore, flow_type: this.dataSource.flowType, }) + for (const eventType of this.dataSource.eventTypes) { + this.bufferStats.set(eventType, { eventCount: 0, eventReservedCount: 0, eventSizeBytes: 0 }) + } this.bufferState = { timeBucket: format(startOfHour(utc(new Date())), "yyyyMMddHH0000"), eventId: TimeUuid.now().toString(), @@ -124,8 +141,6 @@ export class FlowcoreDataPump { public getSnapshot(): PulseSnapshot | null { if (!this.running) return null - const reserved = this.buffer.filter((b) => b.status === "reserved").length - const sizeBytes = this.buffer.reduce((sum, b) => sum + JSON.stringify(b.event.payload).length, 0) return { pathwayId: this.pulseEmitter ? "" : "", // set by caller flowType: this.dataSource.flowType, @@ -133,8 +148,8 @@ export class FlowcoreDataPump { eventId: this.bufferState.eventId, isLive: this.isLive, bufferDepth: this.buffer.length, - bufferReserved: reserved, - bufferSizeBytes: sizeBytes, + bufferReserved: this.bufferReservedCount, + bufferSizeBytes: this.bufferSizeBytes, acknowledgedTotal: this.acknowledgedCount, failedTotal: this.failedCount, pulledTotal: this.pulledCount, @@ -219,7 +234,7 @@ export class FlowcoreDataPump { this.running = true this.startedAt = Date.now() this.nextCursor = undefined - this.updateMetricsGauges() + this.updateMetricsGauges(true) this.pulseEmitter?.start() const currentState = await this.stateManager.getState() const timeBucket = currentState @@ -293,7 +308,8 @@ export class FlowcoreDataPump { this.processLoopRestartAttempts = 0 this.mainLoopRestartAttempts = 0 this.buffer = [] - this.updateMetricsGauges() + this.resetBufferStats() + this.updateMetricsGauges(true) this.pulseEmitter?.stop() this.abortController?.abort() this.waiterBufferThreshold?.() @@ -346,7 +362,7 @@ export class FlowcoreDataPump { this.pulledCount += events.length this.mainLoopRestartAttempts = 0 - this.buffer.push(...events.map((event) => ({ event, status: "open" as const, deliveryCount: 0 }))) + this.addEventsToBuffer(events) this.nextCursor = nextCursor this.updateMetricsGauges() @@ -425,8 +441,10 @@ export class FlowcoreDataPump { event.status = "reserved" event.deliveryId = deliveryId event.deliveryCount++ + this.updateReservedStats(event, 1) events.push(event.event) - this.incMetricsCounter("pulled", event.event.eventType, JSON.stringify(event.event).length) + event.eventSizeBytes ??= textEncoder.encode(JSON.stringify(event.event)).byteLength + this.incMetricsCounter("pulled", event.event.eventType, event.eventSizeBytes) if (events.length === amount) { break } @@ -441,10 +459,12 @@ export class FlowcoreDataPump { this.updateMetricsGauges() setTimeout(() => { - this.reOpen( + void this.reOpen( events.map((event) => event.eventId), deliveryId, - ) + ).catch((error) => { + this.logger?.error("Failed to reopen events after acknowledgement timeout", { error }) + }) }, this.options.achknowledgeTimeoutMs) return events @@ -458,12 +478,14 @@ export class FlowcoreDataPump { if (!this.running) { return } + const eventIdSet = new Set(eventIds) const checkpointEventId = this.replayObserver.observeAcknowledgement(() => { const lastEventInBuffer = this.buffer[this.buffer.length - 1] this.buffer = this.buffer.filter((event) => { - if (eventIds.includes(event.event.eventId)) { + if (eventIdSet.has(event.event.eventId)) { this.incMetricsCounter("acknowledged", event.event.eventType, 1) this.acknowledgedCount++ + this.removeFromBufferStats(event) return false } return true @@ -476,12 +498,12 @@ export class FlowcoreDataPump { return this.buffer.length ? undefined : lastEventInBuffer?.event.eventId }) - await this.updateState(checkpointEventId) - this.updateMetricsGauges() - if (!this.buffer.length) { - this.waiterBufferEmpty?.() + try { + await this.updateState(checkpointEventId) + } finally { + this.notifyBufferEmpty() } } @@ -490,46 +512,54 @@ export class FlowcoreDataPump { return } const lastEventInBuffer = this.buffer[this.buffer.length - 1] + const eventIdSet = new Set(eventIds) const failedEvents: FlowcoreEvent[] = [] this.buffer = this.buffer.filter((event) => { - if (eventIds.includes(event.event.eventId)) { + if (eventIdSet.has(event.event.eventId)) { this.incMetricsCounter("failed", event.event.eventType, 1) this.failedCount++ failedEvents.push(event.event) + this.removeFromBufferStats(event) return false } return true }) this.logger?.info(`Failed ${failedEvents.length} events`) - void this.options.processor?.failedHandler?.(failedEvents) if (this.buffer.length <= this.options.bufferSize - this.options.bufferThreshold) { this.waiterBufferThreshold?.() } - await this.updateState(this.buffer.length ? undefined : lastEventInBuffer?.event.eventId) + this.updateMetricsGauges() - if (!this.buffer.length) { - this.waiterBufferEmpty?.() + try { + await this.options.processor?.failedHandler?.(failedEvents) + await this.updateState(this.buffer.length ? undefined : lastEventInBuffer?.event.eventId) + } finally { + this.notifyBufferEmpty() } } private async reOpen(eventIds: string[], deliveryId: string) { + const eventIdSet = new Set(eventIds) let lastEvent: FlowcoreEvent | undefined const failedEvents: FlowcoreEvent[] = [] const reopenedEvents: FlowcoreEvent[] = [] this.buffer = this.buffer.filter((event) => { - if (event.deliveryId !== deliveryId || !eventIds.includes(event.event.eventId)) { + if (event.deliveryId !== deliveryId || !eventIdSet.has(event.event.eventId)) { return true } if (this.options.maxRedeliveryCount > -1 && event.deliveryCount > this.options.maxRedeliveryCount) { this.incMetricsCounter("failed", event.event.eventType, 1) + this.failedCount++ failedEvents.push(event.event) lastEvent = event.event + this.removeFromBufferStats(event) return false } event.status = "open" event.deliveryId = undefined + this.updateReservedStats(event, -1) reopenedEvents.push(event.event) return true }) @@ -546,17 +576,25 @@ export class FlowcoreDataPump { } this.logger?.info(`Failed ${failedEvents.length} events`) - void this.options.processor?.failedHandler?.(failedEvents) - void this.finallyFailedHandler?.(failedEvents) if (this.buffer.length <= this.options.bufferSize - this.options.bufferThreshold) { this.waiterBufferThreshold?.() } - await this.updateState(this.buffer.length ? undefined : lastEvent?.eventId) - - if (!this.buffer.length) { - this.waiterBufferEmpty?.() + try { + const callbackResults = await Promise.allSettled([ + Promise.resolve().then(() => this.options.processor?.failedHandler?.(failedEvents)), + Promise.resolve().then(() => this.finallyFailedHandler?.(failedEvents)), + ]) + const callbackFailure = callbackResults.find( + (result): result is PromiseRejectedResult => result.status === "rejected", + ) + if (callbackFailure) { + throw callbackFailure.reason + } + await this.updateState(this.buffer.length ? undefined : lastEvent?.eventId) + } finally { + this.notifyBufferEmpty() } } @@ -598,63 +636,76 @@ export class FlowcoreDataPump { // #region Metrics - private updateMetricsGauges() { - const stats = new Map< - string, - { - eventCount: number - eventReservedCount: number - eventSizeBytes: number + private addEventsToBuffer(events: FlowcoreEvent[]): void { + for (const event of events) { + const item: FlowcoreDataPumpBufferItem = { + event, + status: "open", + deliveryCount: 0, + payloadSizeBytes: textEncoder.encode(JSON.stringify(event.payload)).byteLength, + } + this.buffer.push(item) + this.bufferSizeBytes += item.payloadSizeBytes + const stat = this.bufferStats.get(event.eventType) + if (stat) { + stat.eventCount++ + stat.eventSizeBytes += item.payloadSizeBytes } - >() - for (const eventType of this.dataSource.eventTypes) { - stats.set(eventType, { - eventCount: 0, - eventReservedCount: 0, - eventSizeBytes: 0, - }) } + } - for (const item of this.buffer) { - const stat = stats.get(item.event.eventType) - if (!stat) { - continue - } - stat.eventCount++ - if (item.status === "reserved") { - stat.eventReservedCount++ - } - stat.eventSizeBytes += JSON.stringify(item.event.payload).length + private updateReservedStats(item: FlowcoreDataPumpBufferItem, delta: 1 | -1): void { + this.bufferReservedCount += delta + const stat = this.bufferStats.get(item.event.eventType) + if (stat) stat.eventReservedCount += delta + } + + private removeFromBufferStats(item: FlowcoreDataPumpBufferItem): void { + this.bufferSizeBytes -= item.payloadSizeBytes + const stat = this.bufferStats.get(item.event.eventType) + if (stat) { + stat.eventCount-- + stat.eventSizeBytes -= item.payloadSizeBytes } + if (item.status === "reserved") this.updateReservedStats(item, -1) + } - for (const [eventType, stat] of stats) { - metrics.bufferEventCountGauge.set( - { - tenant: this.dataSource.tenant, - data_core: this.dataSource.dataCore, - flow_type: this.dataSource.flowType, - event_type: eventType, - }, - stat.eventCount, - ) - metrics.bufferReservedEventCountGauge.set( - { - tenant: this.dataSource.tenant, - data_core: this.dataSource.dataCore, - flow_type: this.dataSource.flowType, - event_type: eventType, - }, - stat.eventReservedCount, - ) - metrics.bufferSizeBytesGauge.set( - { - tenant: this.dataSource.tenant, - data_core: this.dataSource.dataCore, - flow_type: this.dataSource.flowType, - event_type: eventType, - }, - stat.eventSizeBytes, - ) + private resetBufferStats(): void { + this.bufferReservedCount = 0 + this.bufferSizeBytes = 0 + for (const stat of this.bufferStats.values()) { + stat.eventCount = 0 + stat.eventReservedCount = 0 + stat.eventSizeBytes = 0 + } + } + + private updateMetricsGauges(synchronous = false): void { + if (synchronous) { + this.gaugePublicationScheduled = false + this.publishMetricsGauges() + return + } + if (this.gaugePublicationScheduled) return + this.gaugePublicationScheduled = true + queueMicrotask(() => { + if (!this.gaugePublicationScheduled) return + this.gaugePublicationScheduled = false + this.publishMetricsGauges() + }) + } + + private publishMetricsGauges(): void { + for (const [eventType, stat] of this.bufferStats) { + const labels = { + tenant: this.dataSource.tenant, + data_core: this.dataSource.dataCore, + flow_type: this.dataSource.flowType, + event_type: eventType, + } + metrics.bufferEventCountGauge.set(labels, stat.eventCount) + metrics.bufferReservedEventCountGauge.set(labels, stat.eventReservedCount) + metrics.bufferSizeBytesGauge.set(labels, stat.eventSizeBytes) } } @@ -717,6 +768,12 @@ export class FlowcoreDataPump { } private waiterBufferEmpty?: () => void + private notifyBufferEmpty(): void { + if (!this.buffer.length) { + this.waiterBufferEmpty?.() + } + } + private async waitForBufferEmpty() { if (!this.buffer.length) { return diff --git a/src/data-pump/data-source.ts b/src/data-pump/data-source.ts index d50347f..e073bef 100644 --- a/src/data-pump/data-source.ts +++ b/src/data-pump/data-source.ts @@ -49,6 +49,8 @@ export class FlowcoreDataSource { protected eventTypeIds?: string[] /** Cached time buckets */ protected timeBuckets?: string[] + /** Index for each cached bucket. */ + private timeBucketIndexes = new Map() /** * Creates a new FlowcoreDataSource instance @@ -205,16 +207,48 @@ export class FlowcoreDataSource { eventTypeId: (await this.getEventTypeIds()) as [string, ...string[]], cursor: cursor || undefined, pageSize: 10_000, + order: "asc", }), this.options.directMode, ) timeBuckets.push(...result.timeBuckets) cursor = result.nextCursor } while (cursor !== undefined) - this.timeBuckets = timeBuckets + const normalizedTimeBuckets = [...new Set(timeBuckets.sort())] + this.timeBuckets = normalizedTimeBuckets + this.timeBucketIndexes = new Map() + for (let index = 0; index < normalizedTimeBuckets.length; index++) { + this.timeBucketIndexes.set(normalizedTimeBuckets[index], index) + } return this.timeBuckets } + private lowerBoundTimeBucket(timeBucket: string): number { + const timeBuckets = this.timeBuckets ?? [] + const target = Number.parseFloat(timeBucket) + let low = 0 + let high = timeBuckets.length + while (low < high) { + const middle = low + Math.floor((high - low) / 2) + if (Number.parseFloat(timeBuckets[middle]) < target) low = middle + 1 + else high = middle + } + return low + } + + private upperBoundTimeBucket(timeBucket: string): number { + const timeBuckets = this.timeBuckets ?? [] + const target = Number.parseFloat(timeBucket) + let low = 0 + let high = timeBuckets.length + while (low < high) { + const middle = low + Math.floor((high - low) / 2) + if (Number.parseFloat(timeBuckets[middle]) <= target) low = middle + 1 + else high = middle + } + return low + } + /** * Gets the next time bucket after the specified time bucket * @param timeBucket - The reference time bucket @@ -226,8 +260,8 @@ export class FlowcoreDataSource { return null } const timeBuckets = await this.getTimeBuckets() - const index = timeBuckets.indexOf(closestTimeBucket) - if (index === -1) { + const index = this.timeBucketIndexes.get(closestTimeBucket) + if (index === undefined) { throw new Error(`Could not get next timeBucket, timeBucket ${timeBucket} not found`) } return timeBuckets[index + 1] ?? null @@ -245,16 +279,14 @@ export class FlowcoreDataSource { if (!timeBucket.match(/^\d{14}$/)) { throw new Error(`Invalid timebucket: ${timeBucket}`) } + if (!timeBuckets.length) { + return null + } if (getBefore) { - return ( - timeBuckets.findLast((t) => Number.parseFloat(t) <= Number.parseFloat(timeBucket)) ?? - timeBuckets[timeBuckets.length - 1] - ) + const index = this.upperBoundTimeBucket(timeBucket) - 1 + return timeBuckets[index] ?? timeBuckets[timeBuckets.length - 1] } - return ( - timeBuckets.find((t) => Number.parseFloat(t) >= Number.parseFloat(timeBucket)) ?? - timeBuckets[timeBuckets.length - 1] - ) + return timeBuckets[this.lowerBoundTimeBucket(timeBucket)] ?? timeBuckets[timeBuckets.length - 1] } /** diff --git a/test/tests/data-pump-low-risk.test.ts b/test/tests/data-pump-low-risk.test.ts new file mode 100644 index 0000000..428e847 --- /dev/null +++ b/test/tests/data-pump-low-risk.test.ts @@ -0,0 +1,504 @@ +import { afterEach, describe, expect, it } from "bun:test" +import type { EventListOutput, FlowcoreEvent } from "@flowcore/sdk" +import { TimeUuid } from "@flowcore/time-uuid" +import { metrics } from "../../src/data-pump/metrics.ts" +import { FlowcoreDataPump } from "../../src/data-pump/data-pump.ts" +import { FlowcoreDataSource } from "../../src/data-pump/data-source.ts" +import type { FlowcoreLogger } from "../../src/data-pump/types.ts" + +const pumps: FlowcoreDataPump[] = [] + +afterEach(() => { + for (const pump of pumps.splice(0)) pump.stop() +}) + +function makeEvent(eventType: string, payload: Record, offsetMs = 0): FlowcoreEvent { + return { + eventId: TimeUuid.fromDate(new Date(Date.now() + offsetMs)).toString(), + timeBucket: "20260804120000", + tenant: "test-tenant", + dataCoreId: "test-data-core", + flowType: "test-flow-type", + eventType, + metadata: {}, + validTime: new Date().toISOString(), + payload, + } +} + +class OnePageDataSource extends FlowcoreDataSource { + private fetched = false + + constructor(private readonly page: FlowcoreEvent[]) { + super({ + auth: { apiKey: "fc_testid_testsecret" }, + dataSource: { + tenant: "test-tenant", + dataCore: "test-data-core", + flowType: "test-flow-type", + eventTypes: ["alpha.0", "beta.0"], + }, + noTranslation: true, + }) + } + + public override getClosestTimeBucket(): Promise { + return Promise.resolve("20260804120000") + } + + public override getNextTimeBucket(): Promise { + return Promise.resolve(null) + } + + public override getEvents(): Promise { + if (this.fetched) return new Promise(() => {}) + this.fetched = true + return Promise.resolve({ events: this.page, nextCursor: undefined }) + } +} + +async function createStartedPump( + events: FlowcoreEvent[], + maxRedeliveryCount = 3, + setState: (state: { timeBucket: string; eventId?: string }) => Promise | void = () => {}, + logger?: FlowcoreLogger, + achknowledgeTimeoutMs = 60_000, +): Promise { + const source = new OnePageDataSource(events) + const pump = FlowcoreDataPump.create( + { + auth: { apiKey: "fc_testid_testsecret" }, + dataSource: { + tenant: "test-tenant", + dataCore: "test-data-core", + flowType: "test-flow-type", + eventTypes: ["alpha.0", "beta.0"], + }, + stateManager: { getState: () => null, setState }, + notifier: { type: "poller", intervalMs: 60_000 }, + bufferSize: Math.max(events.length, 1), + achknowledgeTimeoutMs, + maxRedeliveryCount, + noTranslation: true, + logger, + }, + source, + ) + pumps.push(pump) + void pump.start(() => {}) + for (let attempt = 0; attempt < 100 && pump.getSnapshot()?.bufferDepth !== events.length; attempt++) { + await new Promise((resolve) => setTimeout(resolve, 1)) + } + expect(pump.getSnapshot()?.bufferDepth).toBe(events.length) + return pump +} + +async function expectWaiterToResolve(waiter: Promise): Promise { + const result = await Promise.race([ + waiter.then(() => "resolved" as const), + new Promise<"timed out">((resolve) => setTimeout(() => resolve("timed out"), 20)), + ]) + expect(result).toBe("resolved") +} + +async function gaugeValue(gauge: typeof metrics.bufferEventCountGauge, eventType: string): Promise { + await Promise.resolve() + const result = await gauge.get() + return result.values.find( + (value) => + value.labels.tenant === "test-tenant" && + value.labels.data_core === "test-data-core" && + value.labels.flow_type === "test-flow-type" && + value.labels.event_type === eventType, + )?.value +} + +async function pulledBytesValue(eventType: string): Promise { + const result = await metrics.eventsPulledSizeBytesCounter.get() + return ( + result.values.find( + (value) => + value.labels.tenant === "test-tenant" && + value.labels.data_core === "test-data-core" && + value.labels.flow_type === "test-flow-type" && + value.labels.event_type === eventType, + )?.value ?? 0 + ) +} + +function idsWithoutIncludes(ids: string[]): string[] { + Object.defineProperty(ids, "includes", { + value: () => { + throw new Error("linear membership lookup used") + }, + }) + return ids +} + +function setFailedHandler(pump: FlowcoreDataPump, failedHandler: () => Promise): void { + const internals = pump as unknown as { + options: { + processor?: { handler: (events: FlowcoreEvent[]) => Promise; failedHandler?: () => Promise } + } + } + internals.options.processor = { handler: async () => {}, failedHandler } +} + +describe("data pump low-risk buffer bookkeeping", () => { + it("coalesces repeated gauge publications into one microtask", async () => { + const pump = await createStartedPump([]) + const internals = pump as unknown as { + publishMetricsGauges: () => void + updateMetricsGauges: () => void + } + const publish = internals.publishMetricsGauges.bind(pump) + let publications = 0 + internals.publishMetricsGauges = () => { + publications++ + publish() + } + + internals.updateMetricsGauges() + internals.updateMetricsGauges() + internals.updateMetricsGauges() + expect(publications).toBe(0) + await Promise.resolve() + expect(publications).toBe(1) + }) + + it("caches payload and event encodings once and keeps pulse/gauge snapshots equivalent through transitions", async () => { + let serializations = 0 + const payload = { + value: "payload", + toJSON() { + serializations++ + return { value: this.value } + }, + } + const event = makeEvent("alpha.0", payload) + const expectedPayloadBytes = JSON.stringify({ value: "payload" }).length + const pump = await createStartedPump([event]) + + expect(serializations).toBe(1) + expect(pump.getSnapshot()).toMatchObject({ + bufferDepth: 1, + bufferReserved: 0, + bufferSizeBytes: expectedPayloadBytes, + }) + expect(await gaugeValue(metrics.bufferEventCountGauge, "alpha.0")).toBe(1) + expect(await gaugeValue(metrics.bufferReservedEventCountGauge, "alpha.0")).toBe(0) + expect(await gaugeValue(metrics.bufferSizeBytesGauge, "alpha.0")).toBe(expectedPayloadBytes) + + const [reserved] = await pump.reserve(1) + expect(serializations).toBe(2) + expect(pump.getSnapshot()).toMatchObject({ + bufferDepth: 1, + bufferReserved: 1, + bufferSizeBytes: expectedPayloadBytes, + }) + expect(await gaugeValue(metrics.bufferReservedEventCountGauge, "alpha.0")).toBe(1) + + await pump.fail(idsWithoutIncludes([reserved.eventId])) + expect(serializations).toBe(2) + expect(pump.getSnapshot()).toMatchObject({ bufferDepth: 0, bufferReserved: 0, bufferSizeBytes: 0 }) + expect(await gaugeValue(metrics.bufferEventCountGauge, "alpha.0")).toBe(0) + expect(await gaugeValue(metrics.bufferReservedEventCountGauge, "alpha.0")).toBe(0) + expect(await gaugeValue(metrics.bufferSizeBytesGauge, "alpha.0")).toBe(0) + }) + + it("uses Set membership for acknowledgement and removes only matching IDs", async () => { + const events = [makeEvent("alpha.0", { index: 0 }, 0), makeEvent("beta.0", { index: 1 }, 1)] + const pump = await createStartedPump(events) + await pump.reserve(2) + + await pump.acknowledge(idsWithoutIncludes([events[0].eventId])) + + expect(pump.getSnapshot()).toMatchObject({ bufferDepth: 1, bufferReserved: 1, acknowledgedTotal: 1 }) + expect(await gaugeValue(metrics.bufferEventCountGauge, "alpha.0")).toBe(0) + expect(await gaugeValue(metrics.bufferEventCountGauge, "beta.0")).toBe(1) + }) + + it("uses Set membership when reopening and refreshes reserved gauges", async () => { + const event = makeEvent("alpha.0", { index: 0 }) + const pump = await createStartedPump([event]) + await pump.reserve(1) + const deliveryId = (pump as unknown as { buffer: Array<{ deliveryId?: string }> }).buffer[0].deliveryId! + + await (pump as unknown as { reOpen: (ids: string[], deliveryId: string) => Promise }).reOpen( + idsWithoutIncludes([event.eventId]), + deliveryId, + ) + + expect(pump.getSnapshot()).toMatchObject({ bufferDepth: 1, bufferReserved: 0 }) + expect(await gaugeValue(metrics.bufferReservedEventCountGauge, "alpha.0")).toBe(0) + }) + + it("refreshes gauges when timeout reopening finally fails an event", async () => { + const event = makeEvent("alpha.0", { index: 0 }) + const pump = await createStartedPump([event], 0) + await pump.reserve(1) + const deliveryId = (pump as unknown as { buffer: Array<{ deliveryId?: string }> }).buffer[0].deliveryId! + + await (pump as unknown as { reOpen: (ids: string[], deliveryId: string) => Promise }).reOpen( + idsWithoutIncludes([event.eventId]), + deliveryId, + ) + + expect(pump.getSnapshot()).toMatchObject({ bufferDepth: 0, bufferReserved: 0, bufferSizeBytes: 0 }) + expect(await gaugeValue(metrics.bufferEventCountGauge, "alpha.0")).toBe(0) + expect(await gaugeValue(metrics.bufferReservedEventCountGauge, "alpha.0")).toBe(0) + expect(await gaugeValue(metrics.bufferSizeBytesGauge, "alpha.0")).toBe(0) + }) + + it("counts each terminal timeout failure once when requested IDs contain duplicates", async () => { + const event = makeEvent("alpha.0", { index: 0 }) + const pump = await createStartedPump([event], 0) + await pump.reserve(1) + const deliveryId = (pump as unknown as { buffer: Array<{ deliveryId?: string }> }).buffer[0].deliveryId! + + await (pump as unknown as { reOpen: (ids: string[], deliveryId: string) => Promise }).reOpen( + [event.eventId, event.eventId], + deliveryId, + ) + + expect(pump.getSnapshot()).toMatchObject({ bufferDepth: 0, failedTotal: 1 }) + }) + + it("publishes acknowledgement gauges while checkpoint persistence is delayed", async () => { + let releaseCheckpoint!: () => void + const checkpoint = new Promise((resolve) => { + releaseCheckpoint = resolve + }) + const event = makeEvent("alpha.0", { index: 0 }) + const pump = await createStartedPump([event], 3, () => checkpoint) + await pump.reserve(1) + + const acknowledgement = pump.acknowledge([event.eventId]) + + expect(pump.getSnapshot()).toMatchObject({ bufferDepth: 0, bufferReserved: 0 }) + expect(await gaugeValue(metrics.bufferEventCountGauge, "alpha.0")).toBe(0) + expect(await gaugeValue(metrics.bufferReservedEventCountGauge, "alpha.0")).toBe(0) + releaseCheckpoint() + await acknowledgement + }) + + it("publishes failure gauges even when checkpoint persistence rejects", async () => { + const event = makeEvent("alpha.0", { index: 0 }) + const pump = await createStartedPump([event], 3, () => Promise.reject(new Error("checkpoint rejected"))) + await pump.reserve(1) + + await expect(pump.fail([event.eventId])).rejects.toThrow("checkpoint rejected") + + expect(pump.getSnapshot()).toMatchObject({ bufferDepth: 0, bufferReserved: 0 }) + expect(await gaugeValue(metrics.bufferEventCountGauge, "alpha.0")).toBe(0) + expect(await gaugeValue(metrics.bufferReservedEventCountGauge, "alpha.0")).toBe(0) + expect(await gaugeValue(metrics.bufferSizeBytesGauge, "alpha.0")).toBe(0) + }) + + it("wakes the buffer-empty waiter when acknowledgement checkpoint persistence rejects", async () => { + const event = makeEvent("alpha.0", { index: 0 }) + const pump = await createStartedPump([event], 3, () => Promise.reject(new Error("checkpoint rejected"))) + await pump.reserve(1) + const waiter = (pump as unknown as { waitForBufferEmpty: () => Promise }).waitForBufferEmpty() + + await expect(pump.acknowledge([event.eventId])).rejects.toThrow("checkpoint rejected") + await expectWaiterToResolve(waiter) + }) + + it("wakes the buffer-empty waiter when failure checkpoint persistence rejects", async () => { + const event = makeEvent("alpha.0", { index: 0 }) + const pump = await createStartedPump([event], 3, () => Promise.reject(new Error("checkpoint rejected"))) + await pump.reserve(1) + const waiter = (pump as unknown as { waitForBufferEmpty: () => Promise }).waitForBufferEmpty() + + await expect(pump.fail([event.eventId])).rejects.toThrow("checkpoint rejected") + await expectWaiterToResolve(waiter) + }) + + it("awaits and propagates a rejecting failedHandler on direct failure without checkpointing", async () => { + const event = makeEvent("alpha.0", { index: 0 }) + const checkpoints: Array<{ timeBucket: string; eventId?: string }> = [] + const pump = await createStartedPump( + [event], + 3, + (state) => { + checkpoints.push(state) + }, + undefined, + 60_000, + ) + await pump.reserve(1) + setFailedHandler(pump, () => Promise.reject(new Error("failed handler rejected"))) + const waiter = (pump as unknown as { waitForBufferEmpty: () => Promise }).waitForBufferEmpty() + + await expect(pump.fail([event.eventId])).rejects.toThrow("failed handler rejected") + + expect(checkpoints).toEqual([]) + expect(pump.getSnapshot()).toMatchObject({ bufferDepth: 0, failedTotal: 1 }) + await expectWaiterToResolve(waiter) + }) + + it("wakes the buffer-empty waiter and propagates terminal reOpen checkpoint rejection", async () => { + const event = makeEvent("alpha.0", { index: 0 }) + const pump = await createStartedPump([event], 0, () => Promise.reject(new Error("checkpoint rejected"))) + await pump.reserve(1) + const internals = pump as unknown as { + buffer: Array<{ deliveryId?: string }> + reOpen: (ids: string[], deliveryId: string) => Promise + waitForBufferEmpty: () => Promise + } + const waiter = internals.waitForBufferEmpty() + + await expect(internals.reOpen([event.eventId], internals.buffer[0].deliveryId!)).rejects.toThrow( + "checkpoint rejected", + ) + await expectWaiterToResolve(waiter) + }) + + it("logs timer-driven terminal reOpen checkpoint rejection instead of leaving it unhandled", async () => { + const errors: Array<{ message: string | Error; metadata?: Record }> = [] + const logger: FlowcoreLogger = { + debug: () => {}, + info: () => {}, + warn: () => {}, + error: (message, metadata) => errors.push({ message, metadata }), + } + const event = makeEvent("alpha.0", { index: 0 }) + const pump = await createStartedPump([event], 0, () => Promise.reject(new Error("checkpoint rejected")), logger, 1) + + await pump.reserve(1) + for (let attempt = 0; attempt < 100 && !errors.length; attempt++) { + await new Promise((resolve) => setTimeout(resolve, 1)) + } + + expect(errors).toEqual([ + { + message: "Failed to reopen events after acknowledgement timeout", + metadata: { error: expect.any(Error) }, + }, + ]) + expect((errors[0].metadata?.error as Error).message).toBe("checkpoint rejected") + }) + + it("awaits and handles a rejecting failedHandler during timer-driven terminal reOpen", async () => { + const errors: Array<{ message: string | Error; metadata?: Record }> = [] + const checkpoints: Array<{ timeBucket: string; eventId?: string }> = [] + const logger: FlowcoreLogger = { + debug: () => {}, + info: () => {}, + warn: () => {}, + error: (message, metadata) => errors.push({ message, metadata }), + } + const event = makeEvent("alpha.0", { index: 0 }) + const pump = await createStartedPump( + [event], + 0, + (state) => { + checkpoints.push(state) + }, + logger, + 1, + ) + setFailedHandler(pump, () => Promise.reject(new Error("failed handler rejected"))) + const waiter = (pump as unknown as { waitForBufferEmpty: () => Promise }).waitForBufferEmpty() + + await pump.reserve(1) + for (let attempt = 0; attempt < 100 && !errors.length; attempt++) { + await new Promise((resolve) => setTimeout(resolve, 1)) + } + + expect((errors[0].metadata?.error as Error).message).toBe("failed handler rejected") + expect(checkpoints).toEqual([]) + expect(pump.getSnapshot()).toMatchObject({ bufferDepth: 0, failedTotal: 1 }) + await expectWaiterToResolve(waiter) + }) + + it("awaits and handles a rejecting onFinalyFailed callback during timer-driven terminal reOpen", async () => { + const errors: Array<{ message: string | Error; metadata?: Record }> = [] + const checkpoints: Array<{ timeBucket: string; eventId?: string }> = [] + const logger: FlowcoreLogger = { + debug: () => {}, + info: () => {}, + warn: () => {}, + error: (message, metadata) => errors.push({ message, metadata }), + } + const event = makeEvent("alpha.0", { index: 0 }) + const pump = await createStartedPump( + [event], + 0, + (state) => { + checkpoints.push(state) + }, + logger, + 1, + ) + pump.onFinalyFailed(() => Promise.reject(new Error("finally failed handler rejected"))) + const waiter = (pump as unknown as { waitForBufferEmpty: () => Promise }).waitForBufferEmpty() + + await pump.reserve(1) + for (let attempt = 0; attempt < 100 && !errors.length; attempt++) { + await new Promise((resolve) => setTimeout(resolve, 1)) + } + + expect((errors[0].metadata?.error as Error).message).toBe("finally failed handler rejected") + expect(checkpoints).toEqual([]) + expect(pump.getSnapshot()).toMatchObject({ bufferDepth: 0, failedTotal: 1 }) + await expectWaiterToResolve(waiter) + }) + + it("delivers both terminal failure callbacks when failedHandler rejects without an unhandled rejection", async () => { + const errors: Array<{ message: string | Error; metadata?: Record }> = [] + const checkpoints: Array<{ timeBucket: string; eventId?: string }> = [] + const callbacks: string[] = [] + const logger: FlowcoreLogger = { + debug: () => {}, + info: () => {}, + warn: () => {}, + error: (message, metadata) => errors.push({ message, metadata }), + } + const event = makeEvent("alpha.0", { index: 0 }) + const pump = await createStartedPump( + [event], + 0, + (state) => { + checkpoints.push(state) + }, + logger, + 1, + ) + setFailedHandler(pump, async () => { + callbacks.push("failedHandler") + throw new Error("failed handler rejected") + }) + pump.onFinalyFailed(() => { + callbacks.push("onFinalyFailed") + }) + const waiter = (pump as unknown as { waitForBufferEmpty: () => Promise }).waitForBufferEmpty() + + await pump.reserve(1) + for (let attempt = 0; attempt < 100 && !errors.length; attempt++) { + await new Promise((resolve) => setTimeout(resolve, 1)) + } + + expect(callbacks).toEqual(["failedHandler", "onFinalyFailed"]) + expect((errors[0].metadata?.error as Error).message).toBe("failed handler rejected") + expect(checkpoints).toEqual([]) + expect(pump.getSnapshot()).toMatchObject({ bufferDepth: 0, failedTotal: 1 }) + await expectWaiterToResolve(waiter) + }) + + it("measures payload and pulled event metrics in UTF-8 bytes", async () => { + const event = makeEvent("alpha.0", { value: "Halló 👋" }) + const expectedPayloadBytes = new TextEncoder().encode(JSON.stringify(event.payload)).byteLength + const expectedEventBytes = new TextEncoder().encode(JSON.stringify(event)).byteLength + const pulledBytesBefore = await pulledBytesValue("alpha.0") + const pump = await createStartedPump([event]) + + expect(pump.getSnapshot()).toMatchObject({ bufferSizeBytes: expectedPayloadBytes }) + expect(await gaugeValue(metrics.bufferSizeBytesGauge, "alpha.0")).toBe(expectedPayloadBytes) + + await pump.reserve(1) + + expect((await pulledBytesValue("alpha.0")) - pulledBytesBefore).toBe(expectedEventBytes) + }) +}) diff --git a/test/tests/data-source-buckets.test.ts b/test/tests/data-source-buckets.test.ts new file mode 100644 index 0000000..0ab6869 --- /dev/null +++ b/test/tests/data-source-buckets.test.ts @@ -0,0 +1,147 @@ +import { describe, expect, it } from "bun:test" +import { FlowcoreDataSource } from "../../src/data-pump/data-source.ts" + +function createSource(timeBuckets: string[]): FlowcoreDataSource { + const source = new FlowcoreDataSource({ + auth: { apiKey: "fc_testid_testsecret" }, + dataSource: { + tenant: "test-tenant", + dataCore: "test-data-core", + flowType: "test-flow-type", + eventTypes: ["alpha.0"], + }, + noTranslation: true, + }) + ;(source as unknown as { flowcoreClient: { execute: () => Promise } }).flowcoreClient = { + execute: () => Promise.resolve({ timeBuckets, nextCursor: undefined }), + } + return source +} + +function createPagedSource(pages: Array<{ timeBuckets: string[]; nextCursor?: number }>): { + source: FlowcoreDataSource + commandInputs: Array<{ order?: string; cursor?: number }> +} { + const source = createSource([]) + const commandInputs: Array<{ order?: string; cursor?: number }> = [] + let pageIndex = 0 + ;(source as unknown as { flowcoreClient: { execute: (command: unknown) => Promise } }).flowcoreClient = { + execute: (command) => { + commandInputs.push((command as { input: { order?: string; cursor?: number } }).input) + return Promise.resolve(pages[pageIndex++]) + }, + } + return { source, commandInputs } +} + +function currentClosestOracle(timeBuckets: string[], target: string, getBefore: boolean): string | null { + if (!timeBuckets.length) return null + const targetNumber = Number.parseFloat(target) + if (getBefore) { + return timeBuckets.findLast((bucket) => Number.parseFloat(bucket) <= targetNumber) ?? timeBuckets.at(-1)! + } + return timeBuckets.find((bucket) => Number.parseFloat(bucket) >= targetNumber) ?? timeBuckets.at(-1)! +} + +function currentNextOracle(timeBuckets: string[], target: string): string | null { + const uniqueTimeBuckets = [...new Set(timeBuckets)] + const closest = currentClosestOracle(uniqueTimeBuckets, target, false) + if (!closest) return null + return uniqueTimeBuckets[uniqueTimeBuckets.indexOf(closest) + 1] ?? null +} + +describe("time bucket indexed traversal", () => { + it("matches closest semantics and advances strictly past duplicate buckets", async () => { + const sequences = [ + [], + ["20260101000000"], + ["20260101000000", "20260101010000", "20260101020000"], + ["20260101000000", "20260101010000", "20260101010000", "20260101030000"], + ] + const targets = ["20251231230000", "20260101000000", "20260101003000", "20260101010000", "20260101040000"] + + for (const timeBuckets of sequences) { + const source = createSource(timeBuckets) + for (const target of targets) { + expect(await source.getClosestTimeBucket(target)).toBe(currentClosestOracle(timeBuckets, target, false)) + expect(await source.getClosestTimeBucket(target, true)).toBe(currentClosestOracle(timeBuckets, target, true)) + expect(await source.getNextTimeBucket(target)).toBe(currentNextOracle(timeBuckets, target)) + } + } + }) + + it("does not use repeated linear array helpers after building the bucket indexes", async () => { + const source = createSource(["20260101000000", "20260101010000", "20260101010000", "20260101030000"]) + const buckets = await source.getTimeBuckets() + Object.defineProperties(buckets, { + find: { + value: () => { + throw new Error("linear find used") + }, + }, + findLast: { + value: () => { + throw new Error("linear findLast used") + }, + }, + indexOf: { + value: () => { + throw new Error("linear indexOf used") + }, + }, + }) + + expect(await source.getClosestTimeBucket("20260101003000")).toBe("20260101010000") + expect(await source.getClosestTimeBucket("20260101020000", true)).toBe("20260101010000") + expect(await source.getNextTimeBucket("20260101010000")).toBe("20260101030000") + }) + + it("rebuilds lookup state when buckets are force-refreshed", async () => { + const source = createSource(["20260101000000", "20260101010000"]) + expect(await source.getNextTimeBucket("20260101000000")).toBe("20260101010000") + ;(source as unknown as { flowcoreClient: { execute: () => Promise } }).flowcoreClient = { + execute: () => Promise.resolve({ timeBuckets: ["20260101000000", "20260101030000"], nextCursor: undefined }), + } + await source.getTimeBuckets(true) + + expect(await source.getNextTimeBucket("20260101000000")).toBe("20260101030000") + }) + + it("requests ascending pages and normalizes descending or out-of-order buckets", async () => { + const { source, commandInputs } = createPagedSource([ + { + timeBuckets: ["20260101030000", "20260101010000", "20260101020000"], + }, + ]) + + expect(await source.getTimeBuckets()).toEqual(["20260101010000", "20260101020000", "20260101030000"]) + expect(commandInputs.map(({ order }) => order)).toEqual(["asc"]) + expect(await source.getClosestTimeBucket("20260101013000")).toBe("20260101020000") + expect(await source.getNextTimeBucket("20260101010000")).toBe("20260101020000") + }) + + it("deduplicates the complete cross-page catalog and makes forward progress", async () => { + const { source, commandInputs } = createPagedSource([ + { + timeBuckets: ["20260101030000", "20260101010000"], + nextCursor: 2, + }, + { + timeBuckets: ["20260101020000", "20260101010000", "20260101040000"], + }, + ]) + + expect(await source.getTimeBuckets()).toEqual([ + "20260101010000", + "20260101020000", + "20260101030000", + "20260101040000", + ]) + expect(commandInputs.map(({ order, cursor }) => ({ order, cursor }))).toEqual([ + { order: "asc", cursor: undefined }, + { order: "asc", cursor: 2 }, + ]) + expect(await source.getNextTimeBucket("20260101010000")).toBe("20260101020000") + expect(await source.getClosestTimeBucket("20260101025000", true)).toBe("20260101020000") + }) +})