diff --git a/projects/ngx-ui-watchtower/DESIGN.md b/projects/ngx-ui-watchtower/DESIGN.md index 1c0d2df..f030539 100644 --- a/projects/ngx-ui-watchtower/DESIGN.md +++ b/projects/ngx-ui-watchtower/DESIGN.md @@ -614,8 +614,11 @@ all along — use the named method instead. For Observable-driven journeys, the pipeable `traceScenario` operator completes or fails the scenario based on how the stream ends. If it is torn down some other way first — a superseding `switchMap`, `takeUntilDestroyed()`, -a manual unsubscribe — it cancels the scenario instead of leaving it to -settle as a `timeout`. +a manual unsubscribe — what it records depends on whether a value got +through. Before any value, it cancels the scenario instead of leaving it to +settle as a `timeout`. After a value, it completes the scenario, because that +is how `firstValueFrom`, `take(1)` and `first()` end a stream once they have +what they need. ```ts import { traceScenario } from '@absaoss-cps/ngx-ui-watchtower'; @@ -650,13 +653,36 @@ this.searchSubject$.pipe( ``` How it decides: `traceScenario` settles on `complete` with optional derived -outcome metadata, and fails with the caught error on `error`. A teardown that reaches neither — a superseding `switchMap`, `takeUntilDestroyed()`, -a manual unsubscribe — cancels the scenario instead, guarded by `isSettled` -so it never re-cancels one that already completed or failed (`tap`'s own -`unsubscribe` hook fires after every teardown, settled or not). Without -this, a cancelled-by-unsubscription scenario would sit active until its own -timeout and record as `timeout` rather than the caller-driven abandonment -it actually was. +outcome metadata, and fails with the caught error on `error`. A teardown that +reaches neither — a superseding `switchMap`, `takeUntilDestroyed()`, a manual +unsubscribe — settles it by whether the stream has emitted: + +- **No value yet:** cancels the scenario. Without this, a + cancelled-by-unsubscription scenario would sit active until its own timeout + and record as `timeout` rather than the caller-driven abandonment it + actually was. +- **After a value:** completes the scenario, with the outcome mapped from the + last value. `firstValueFrom`, `take(1)` and `first()` unsubscribe upstream + as soon as they have a value, before the source's `complete` arrives, so + without this a successful request would record as `abandoned`. + +Both are guarded by `isSettled`, so a teardown never re-settles a scenario +that already completed or failed (`tap`'s own `unsubscribe` hook fires after +every teardown, settled or not). + +For a multi-value stream — polling, progress, a websocket — stopping after a +value usually means the user gave up, not that the journey succeeded. Opt +back into cancelling with `valueThenTeardown: 'cancel'`: + +```ts +this.api + .pollJobStatus(jobId) + .pipe( + traceScenario(scenario, { valueThenTeardown: 'cancel' }), + takeUntilDestroyed(this.destroyRef) + ) + .subscribe(); +``` ### Measuring what the user waited for diff --git a/projects/ngx-ui-watchtower/src/lib/scenario/uwt-scenario-operators/uwt-scenario-operators.spec.ts b/projects/ngx-ui-watchtower/src/lib/scenario/uwt-scenario-operators/uwt-scenario-operators.spec.ts index 7ec1c38..582564a 100644 --- a/projects/ngx-ui-watchtower/src/lib/scenario/uwt-scenario-operators/uwt-scenario-operators.spec.ts +++ b/projects/ngx-ui-watchtower/src/lib/scenario/uwt-scenario-operators/uwt-scenario-operators.spec.ts @@ -1,4 +1,21 @@ -import { defer, EMPTY, of, Subject, switchMap, throwError } from 'rxjs'; +import { + defer, + EMPTY, + first, + type Observable, + firstValueFrom, + lastValueFrom, + of, + retry, + Subject, + switchMap, + take, + throwError +} from 'rxjs'; +import { UWT_DEFAULT_TELEMETRY_CONFIG } from '../../config/uwt-telemetry-common.config/uwt-telemetry-common.config'; +import { UwtScenarioRecord } from '../../models/uwt-scenario.models/uwt-scenario.models'; +import { UwtNoopTelemetrySink } from '../../sinks/uwt-telemetry/uwt-noop-telemetry.sink/uwt-noop-telemetry.sink'; +import { UWT_DEFAULT_REDACT_CONFIG } from '../../utils/uwt-telemetry-redact.util/uwt-telemetry-redact.util'; import { UwtScenario } from '../uwt-scenario/uwt-scenario'; import { traceScenario } from './uwt-scenario-operators'; @@ -292,4 +309,202 @@ describe('traceScenario operator', () => { }); }); }); + + describe('with a real scenario', () => { + let settled: UwtScenarioRecord[]; + + const countRows = (rows: string[]) => ({ + metadata: { count: rows.length } + }); + + function createScenario(name = 'load-customers'): UwtScenario { + return new UwtScenario( + { name }, + { + identity: { + application: 'test-app', + environment: 'test', + version: '1.0.0' + }, + scenarioTelemetryConfig: { + ...UWT_DEFAULT_TELEMETRY_CONFIG.scenario + }, + redact: UWT_DEFAULT_REDACT_CONFIG, + sink: new UwtNoopTelemetrySink(), + onSettled: (_id, record) => settled.push(record) + } + ); + } + + beforeEach(() => { + settled = []; + }); + + describe('single-value sources', () => { + it('should settle as success with subscribe()', () => { + const real = createScenario(); + + of(['a', 'b']).pipe(traceScenario(real, countRows)).subscribe(); + + expect(real.status).toBe('success'); + expect(settled[0].metadata).toEqual({ count: 2 }); + }); + + it('should settle as success with lastValueFrom', async () => { + const real = createScenario(); + + await lastValueFrom( + of(['a', 'b']).pipe(traceScenario(real, countRows)) + ); + + expect(real.status).toBe('success'); + expect(settled[0].metadata).toEqual({ count: 2 }); + }); + + it('should settle as success with firstValueFrom', async () => { + const real = createScenario(); + + const rows = await firstValueFrom( + of(['a', 'b']).pipe(traceScenario(real, countRows)) + ); + + expect(rows).toEqual(['a', 'b']); + expect(real.status).toBe('success'); + expect(settled[0].metadata).toEqual({ count: 2 }); + }); + + it('should settle as success with take(1) after the operator', () => { + const real = createScenario(); + + of(['a', 'b']) + .pipe(traceScenario(real, countRows), take(1)) + .subscribe(); + + expect(real.status).toBe('success'); + expect(settled[0].metadata).toEqual({ count: 2 }); + }); + + it('should settle as success with first() after the operator', () => { + const real = createScenario(); + + of(['a', 'b']) + .pipe(traceScenario(real, countRows), first()) + .subscribe(); + + expect(real.status).toBe('success'); + expect(settled[0].metadata).toEqual({ count: 2 }); + }); + + it('should settle as success with take(1) before the operator', () => { + const real = createScenario(); + + of(['a', 'b']) + .pipe(take(1), traceScenario(real, countRows)) + .subscribe(); + + expect(real.status).toBe('success'); + expect(settled[0].metadata).toEqual({ count: 2 }); + }); + + it('should settle as success with take(1) before the operator under firstValueFrom', async () => { + const real = createScenario(); + + await firstValueFrom( + of(['a', 'b']).pipe(take(1), traceScenario(real, countRows)) + ); + + expect(real.status).toBe('success'); + expect(settled[0].metadata).toEqual({ count: 2 }); + }); + }); + + describe('switchMap superseding an inner stream', () => { + it('should abandon a stream superseded before it emitted', () => { + const superseded = createScenario('superseded'); + const latest = createScenario('latest'); + const pendingFirstRequest = new Subject(); + const trigger = new Subject(); + + trigger + .pipe( + switchMap((n) => + n === 1 + ? pendingFirstRequest.pipe(traceScenario(superseded)) + : of(['b']).pipe(traceScenario(latest)) + ) + ) + .subscribe(); + + trigger.next(1); + trigger.next(2); + + expect(superseded.status).toBe('abandoned'); + expect(latest.status).toBe('success'); + }); + + it('should settle as success a stream superseded after it emitted', () => { + const superseded = createScenario('superseded'); + const latest = createScenario('latest'); + const firstRequest = new Subject(); + const trigger = new Subject(); + + trigger + .pipe( + switchMap((n) => + n === 1 + ? firstRequest.pipe(traceScenario(superseded, countRows)) + : of(['b']).pipe(traceScenario(latest)) + ) + ) + .subscribe(); + + trigger.next(1); + firstRequest.next(['a']); // emitted, but never completed + trigger.next(2); + + expect(superseded.status).toBe('success'); + expect(superseded.toRecord().metadata).toEqual({ count: 1 }); + expect(latest.status).toBe('success'); + }); + }); + + it("should abandon a stream torn down after a value with valueThenTeardown: 'cancel'", () => { + const real = createScenario('poll-status'); + + of(['a'], ['b']) + .pipe(traceScenario(real, { valueThenTeardown: 'cancel' }), take(1)) + .subscribe(); + + expect(real.status).toBe('abandoned'); + }); + + describe('retry placement', () => { + function failOnce(): Observable { + let attempts = 0; + return defer(() => + attempts++ === 0 + ? throwError(() => new Error('transient')) + : of(['a']) + ); + } + + it('should settle as success when retry() comes before the operator', () => { + const real = createScenario(); + + failOnce().pipe(retry(1), traceScenario(real)).subscribe(); + + expect(real.status).toBe('success'); + }); + + it('should settle as failure when retry() comes after the operator, even if the retry succeeds', () => { + const real = createScenario(); + + failOnce().pipe(traceScenario(real), retry(1)).subscribe(); + + // Documented limitation: the first error settles the scenario and + // the operator cannot see the downstream retry. + expect(real.status).toBe('failure'); + }); + }); + }); }); diff --git a/projects/ngx-ui-watchtower/src/lib/scenario/uwt-scenario-operators/uwt-scenario-operators.ts b/projects/ngx-ui-watchtower/src/lib/scenario/uwt-scenario-operators/uwt-scenario-operators.ts index 2b6f9d4..b9343b0 100644 --- a/projects/ngx-ui-watchtower/src/lib/scenario/uwt-scenario-operators/uwt-scenario-operators.ts +++ b/projects/ngx-ui-watchtower/src/lib/scenario/uwt-scenario-operators/uwt-scenario-operators.ts @@ -22,6 +22,14 @@ export interface UwtTraceScenarioOptions { * erroring — e.g. a `switchMap` moving on to a newer value. */ cancelOutcome?: UwtScenarioOutcome | (() => UwtScenarioOutcome | void); + + /** + * What a downstream unsubscribe means once a value has been delivered. + * Default `'complete'`, so `firstValueFrom`, `take(1)` and `first()` settle + * as success. Use `'cancel'` for multi-value streams (polling, progress), + * where stopping after a value is abandonment. + */ + valueThenTeardown?: 'complete' | 'cancel'; } /** @@ -54,45 +62,55 @@ export function traceScenario( typeof options === 'function' ? options : options?.outcome; const cancelOutcome = typeof options === 'function' ? undefined : options?.cancelOutcome; + const valueThenTeardown = + (typeof options === 'function' ? undefined : options?.valueThenTeardown) ?? + 'complete'; return (source: Observable): Observable => defer(() => { let lastValue: T | undefined; let hasValue = false; + const complete = () => { + const outcome = + hasValue && outcomeMapper + ? uwtSafe( + 'scenario.traceOutcome', + () => outcomeMapper(lastValue as T), + undefined + ) + : undefined; + scenario.complete(outcome || undefined); + }; + + const unsubscribe = () => { + if (scenario.isSettled) { + return; + } + // The consumer got a value and stopped listening, as + // firstValueFrom, take(1) and first() do: treat it as completion. + if (hasValue && valueThenTeardown === 'complete') { + complete(); + return; + } + const resolved = + typeof cancelOutcome === 'function' + ? uwtSafe('scenario.traceCancelOutcome', cancelOutcome, undefined) + : cancelOutcome; + scenario.cancel(resolved || undefined); + }; + return source.pipe( tap({ next: (value) => { lastValue = value; hasValue = true; }, - complete: () => { - const outcome = - hasValue && outcomeMapper - ? uwtSafe( - 'scenario.traceOutcome', - () => outcomeMapper(lastValue as T), - undefined - ) - : undefined; - scenario.complete(outcome || undefined); - }, + complete, error: (error: unknown) => { scenario.fail({ error }); }, - unsubscribe: () => { - if (!scenario.isSettled) { - const resolved = - typeof cancelOutcome === 'function' - ? uwtSafe( - 'scenario.traceCancelOutcome', - cancelOutcome, - undefined - ) - : cancelOutcome; - scenario.cancel(resolved || undefined); - } - } + unsubscribe }) ); });