SuavePlan/suaveplan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
suaveplan/openspec/changes/etl-completion/specs/etl/spec.md
admin fbc70f96e6 Initial Commit
2026-05-27 23:29:53 +01:00

11 KiB
Raw Permalink Blame History

Capability Spec Delta: etl (etl-completion)

Package: @suaveplan/etl Status: PROPOSED

Every requirement below carries a **Verified by:** line at authoring time. No exceptions. (See openspec/AGENTS.md Section 13.)

ADDED Requirements

Requirement: PipelineContext shared mutable execution state

The package SHALL expose a PipelineContext<TData> class and a createContext<TData>(options) factory. The context SHALL hold mutable data: TData, accumulate errors: EtlError[] and stepResults: StepResult[] in execution order, expose a typed metadata key/value store (set, get, has, delete), accept an injectable Clock, LoggerLike, and AbortSignal, and SHALL be threaded through every stage of Pipeline.run() so cross-step state is observable. The context SHALL expose on(event, listener) / off(event, listener) / emit(event, payload) for the same lifecycle events as EtlHookMap.

Scenario: metadata round-trip

  • WHEN ctx.set("ingestKey", 42) is called and then ctx.get<number>("ingestKey") is called
  • THEN the result is 42 AND ctx.has("ingestKey") returns true

Scenario: errors and stepResults accumulate per stage

  • WHEN a pipeline runs three stages and the second one throws
  • THEN ctx.stepResults contains one entry for stage 1, ctx.errors contains one entry whose cause is the original throw, AND no entry exists for stage 3 unless batch.continueOnError is true

Scenario: AbortSignal threads to stages

  • WHEN the supplied abortSignal is aborted mid-run
  • THEN the next stage receiving the context SHALL observe ctx.abortSignal.aborted === true and abortable stages SHALL bail out

Scenario: injected Clock stamps metadata

  • WHEN the context is constructed with clock: { now: () => 1000 }
  • THEN ctx.get<number>("startedAt") returns 1000 (the runner stamps startedAt from the injected clock at run start)

Verified by: packages/platform/etl/src/core/context/context.test.ts::"round-trips metadata via set / get / has" AND packages/platform/etl/src/core/context/context.test.ts::"accumulates errors and stepResults across stages" AND packages/platform/etl/src/core/context/context.test.ts::"propagates AbortSignal to downstream stages" AND packages/platform/etl/src/core/context/context.spec-conformance.test.ts::"stamps startedAt from injected Clock"

Requirement: BatchConfig with concurrency, per-item timeout, batch timeout

The package SHALL accept a BatchConfig interface (size, concurrency, perItemTimeoutMs, batchTimeoutMs, continueOnError). Pipeline.run() SHALL honour all five fields. concurrency < 1 SHALL throw EtlError(BATCH_CONCURRENCY_INVALID). size < 1 SHALL throw the same code. Per-item timeout exceedance SHALL throw EtlError(BATCH_ITEM_TIMEOUT). Batch timeout exceedance SHALL throw EtlError(BATCH_TIMEOUT). continueOnError: true SHALL allow the pipeline to skip a failed batch and continue with the next.

Scenario: size chunks input correctly

  • WHEN input has 100 items and batch.size is 25
  • THEN the runner processes exactly 4 batches of 25

Scenario: concurrency runs N batches in parallel

  • WHEN batch.concurrency is 3 and 6 batches are submitted
  • THEN at most 3 batches are in-flight simultaneously (verified via shared counter assertion)

Scenario: per-item timeout rejects slow items

  • WHEN batch.perItemTimeoutMs is 50 and a stage's item handler takes 200ms
  • THEN the runner throws EtlError with code BATCH_ITEM_TIMEOUT and context.itemIndex set

Scenario: batch timeout rejects slow batches

  • WHEN batch.batchTimeoutMs is 100 and a whole batch takes 500ms
  • THEN the runner throws EtlError with code BATCH_TIMEOUT and context.batchIndex set

Scenario: continueOnError skips failed batch

  • WHEN batch 2 of 5 throws and batch.continueOnError is true
  • THEN the runner records the error in ctx.errors, emits batch:failed, and continues with batches 3, 4, 5

Verified by: packages/platform/etl/src/core/pipeline/pipeline.test.ts::"size chunks input correctly" AND packages/platform/etl/src/core/pipeline/pipeline.test.ts::"concurrency runs N batches in parallel" AND packages/platform/etl/src/core/pipeline/pipeline.test.ts::"perItemTimeoutMs rejects slow items with BATCH_ITEM_TIMEOUT" AND packages/platform/etl/src/core/pipeline/pipeline.test.ts::"batchTimeoutMs rejects slow batches with BATCH_TIMEOUT" AND packages/platform/etl/src/core/pipeline/pipeline.test.ts::"continueOnError skips a failed batch and continues"

Requirement: EtlHookMap typed lifecycle event bus

The package SHALL expose EtlHookMap extending HookMap from @suaveplan/events::hooks.types with the keys "pipeline:started", "pipeline:completed", "stage:started", "stage:completed", "stage:failed", "batch:started", "batch:completed", "batch:failed". Pipeline SHALL accept an optional hooks?: HookBus<EtlHookMap> and emit each event in the corresponding lifecycle position. Subscriber throws SHALL NOT abort the pipeline.

Scenario: pipeline:started fires before first stage

  • WHEN the pipeline begins a run()
  • THEN subscribers to "pipeline:started" receive { pipelineName, startedAt } BEFORE the first stage's run is called

Scenario: stage:started / stage:completed bracket each stage

  • WHEN a stage runs successfully
  • THEN "stage:started" fires before its handler and "stage:completed" fires after, with { stageName, durationMs }

Scenario: stage:failed fires on stage error

  • WHEN a stage's handler throws
  • THEN "stage:failed" fires with { stageName, error } (where error is an EtlError wrapping the cause)

Scenario: batch:* fires only when batching is enabled

  • WHEN PipelineOptions.batch is supplied
  • THEN "batch:started", "batch:completed", and "batch:failed" fire per batch transition; when batch is omitted, none of the batch-* events fire

Scenario: subscriber throw does not abort pipeline

  • WHEN a subscriber to "pipeline:started" throws synchronously
  • THEN the pipeline continues to "stage:started" and the rest of the run completes

Verified by: packages/platform/etl/src/types/etl-hooks/etl-hooks.test.ts::"pipeline:started fires before first stage" AND packages/platform/etl/src/types/etl-hooks/etl-hooks.test.ts::"stage:started and stage:completed bracket each stage" AND packages/platform/etl/src/types/etl-hooks/etl-hooks.test.ts::"stage:failed fires on stage error with EtlError" AND packages/platform/etl/src/types/etl-hooks/etl-hooks.test.ts::"batch:* fires only when batching is enabled" AND packages/platform/etl/src/types/etl-hooks/etl-hooks.spec-conformance.test.ts::"subscriber throw does not abort pipeline"

Requirement: utils — applyMappings, getNestedValue, setNestedValue, parallelMap, toBatches, withTimeout

The package SHALL export six pure utilities under utils/utils. Each SHALL have the signatures and behaviors below.

  • applyMappings<TIn, TOut>(record: TIn, mappings: readonly FieldMapping[]): TOut — produces a new object by reading each mapping's from path from record, optionally running transform, and writing the result at the to path.
  • getNestedValue(obj: unknown, path: string): unknown — splits path on . and walks obj; returns undefined for any missing or non-object segment.
  • setNestedValue(obj: Record<string, unknown>, path: string, value: unknown): void — splits path on ., creating intermediate objects as needed, and assigns value at the leaf.
  • parallelMap<T, U>(items: readonly T[], fn: (item: T, index: number) => Promise<U>, concurrency: number): Promise<U[]> — runs fn over items with at most concurrency in-flight; preserves index order in the result.
  • toBatches<T>(items: readonly T[], size: number): T[][] — partitions items into chunks of size; final chunk may be shorter.
  • withTimeout<T>(promise: Promise<T>, ms: number): Promise<T> — rejects with EtlError(TIMEOUT) if promise does not settle within ms.

Scenario: applyMappings copies and transforms

  • WHEN applyMappings({ user: { name: "ada" } }, [{ from: "user.name", to: "displayName", transform: (v) => String(v).toUpperCase() }]) is called
  • THEN the result is { displayName: "ADA" }

Scenario: getNestedValue returns undefined for missing path

  • WHEN getNestedValue({ a: { b: 1 } }, "a.c.d") is called
  • THEN the result is undefined (no throw)

Scenario: parallelMap honours concurrency limit

  • WHEN parallelMap([1,2,3,4,5,6,7,8,9,10], slowAdd, 3) is called
  • THEN at most 3 calls to slowAdd are in-flight at any moment AND the result preserves input order

Scenario: parallelMap rejects on concurrency < 1

  • WHEN parallelMap([1,2,3], fn, 0) is called
  • THEN the call rejects with EtlError(BATCH_CONCURRENCY_INVALID)

Scenario: toBatches partitions exactly

  • WHEN toBatches([1,2,3,4,5,6,7], 3) is called
  • THEN the result is [[1,2,3],[4,5,6],[7]]

Scenario: withTimeout rejects with TIMEOUT code

  • WHEN withTimeout(new Promise(() => {}), 50) is called
  • THEN the call rejects with EtlError whose code is TIMEOUT

Verified by: packages/platform/etl/src/utils/utils.test.ts::"applyMappings copies and transforms field" AND packages/platform/etl/src/utils/utils.test.ts::"getNestedValue returns undefined for missing path" AND packages/platform/etl/src/utils/utils.test.ts::"toBatches partitions input with remainder" AND packages/platform/etl/src/utils/utils.spec-conformance.test.ts::"parallelMap honours concurrency limit empirically" AND packages/platform/etl/src/utils/utils.spec-conformance.test.ts::"parallelMap rejects on concurrency < 1 with BATCH_CONCURRENCY_INVALID" AND packages/platform/etl/src/utils/utils.spec-conformance.test.ts::"withTimeout rejects with EtlError(TIMEOUT)"

Requirement: FieldMapping type and mapTransform mapping integration

The package SHALL export a FieldMapping interface — { from: string; to: string; transform?: (value: unknown) => unknown }. TransformConfig SHALL accept mappings?: readonly FieldMapping[]. mapTransform SHALL apply the mappings (via applyMappings) BEFORE invoking the transform function so the function sees the post-mapping shape.

Scenario: FieldMapping type shape

  • WHEN consumer code declares const m: FieldMapping = { from: "a.b", to: "c", transform: (v) => v }
  • THEN the type-check passes; from and to are string-only; transform is the only optional field

Scenario: mapTransform applies mappings before fn

  • WHEN mapTransform({ mappings: [{ from: "user.name", to: "name" }], fn: (r: { name: string }) => r.name.toUpperCase() }) is run over { user: { name: "ada" } }
  • THEN the output is "ADA" (the transform function received { name: "ada" }, not the original { user: { name: "ada" } })

Verified by: packages/platform/etl/src/types/etl-hooks/etl-hooks.spec-conformance.test.ts::"FieldMapping shape pinned via expectTypeOf" AND packages/platform/etl/src/stages/transform/transform.test.ts::"mapTransform applies mappings before fn"

Reference in New Issue View Git Blame Copy Permalink