From 384580c40f1c4f917e9703c5f5dd7e16d702d632 Mon Sep 17 00:00:00 2001 From: Patrick Roza Date: Fri, 3 Jul 2026 08:42:29 +0200 Subject: [PATCH] Add cached failure invalidation option --- .changeset/cached-failure-invalidation.md | 5 + packages/vue/src/makeClient.ts | 2 +- packages/vue/src/mutate.ts | 123 ++++++++++++++++-- .../vue/test/dependencyInvalidation.test.ts | 95 +++++++++++++- 4 files changed, 214 insertions(+), 11 deletions(-) create mode 100644 .changeset/cached-failure-invalidation.md diff --git a/.changeset/cached-failure-invalidation.md b/.changeset/cached-failure-invalidation.md new file mode 100644 index 000000000..7deeccc07 --- /dev/null +++ b/.changeset/cached-failure-invalidation.md @@ -0,0 +1,5 @@ +--- +"@effect-app/vue": patch +--- + +Add opt-in mutation failure invalidation from the last successful mutation invalidation. diff --git a/packages/vue/src/makeClient.ts b/packages/vue/src/makeClient.ts index 2ed232d71..cb884f2df 100644 --- a/packages/vue/src/makeClient.ts +++ b/packages/vue/src/makeClient.ts @@ -222,7 +222,7 @@ export type StreamFnStreamExtension = Req extends */ export type StreamMutation2WithExtensions = Req extends RequestStreamHandlerWithInput ? - & ((input: I) => Stream.Stream) + & ((input: I, options?: Pick) => Stream.Stream) & { readonly id: Id readonly wrap: ( diff --git a/packages/vue/src/mutate.ts b/packages/vue/src/mutate.ts index c4fcb3efe..42e57688e 100644 --- a/packages/vue/src/mutate.ts +++ b/packages/vue/src/mutate.ts @@ -8,6 +8,7 @@ import * as Option from "effect-app/Option" import { isReadonlyArrayNonEmpty } from "effect/Array" import type * as Cause from "effect/Cause" import * as Exit from "effect/Exit" +import * as Hash from "effect/Hash" import * as Ref from "effect/Ref" import * as Stream from "effect/Stream" import * as AsyncResult from "effect/unstable/reactivity/AsyncResult" @@ -151,6 +152,14 @@ const queryKeyFromFilters = ( return Array.isArray(queryKey) ? queryKey : undefined } +interface SuccessfulInvalidation { + readonly clientKeys: ReadonlyArray> + readonly serverKeys: ReadonlyArray> + readonly writeDependencies: DataDependencies.DataDependencies +} + +const successfulInvalidations = new Map() + export interface MutationOptionsBase { /** * By default we invalidate one level of the query key, e.g $project/$configuration.get, we invalidate $project. @@ -172,6 +181,12 @@ export interface MutationOptionsBase input?: unknown, output?: Exit.Exit ) => InvalidationEntry[] + /** + * On success, remember the invalidation sources for this mutation+input. On a + * later failure for the same mutation+input, replay them so stale query data + * can self-heal even when the failed response carries no write metadata. + */ + invalidateOnFailureFromLastSuccess?: boolean /** * Run an additional Effect after the mutation succeeds. Its output becomes the * final result returned to the caller. Query cache is invalidated once on @@ -265,6 +280,7 @@ export const asStreamResult = ( const buildInvalidateCache = ( self: { id: string; options?: ClientForOptions; disableQueryInvalidation?: boolean }, queryInvalidation: MutationOptionsBase["queryInvalidation"] | undefined, + invalidateOnFailureFromLastSuccess: boolean | undefined, queryInvalidator: QueryInvalidator ) => { // Concrete reactivity keys to invalidate: a raw query key, one derived from an `{ id }` @@ -303,6 +319,44 @@ const buildInvalidateCache = ( return [] } + const uniqueKeys = (keys: ReadonlyArray>): ReadonlyArray> => { + const out: Array> = [] + const seen = new Set() + for (const key of keys) { + const hash = Hash.hash(key) + if (seen.has(hash)) continue + seen.add(hash) + out.push(key) + } + return out + } + + const mutationKey = (input: unknown) => Hash.hash([makeQueryKey(self), input]) + + const rememberSuccess = ( + input: unknown, + output: Exit.Exit, + serverKeys: ReadonlyArray, + writeDependencies: DataDependencies.DataDependencies = DataDependencies.empty() + ) => + Effect.suspend(() => { + if (!invalidateOnFailureFromLastSuccess || self.disableQueryInvalidation || output._tag !== "Success") { + return Effect.void + } + const clientKeys = getClientInvalidationKeys(input, output) + const cacheKey = mutationKey(input) + if ( + isReadonlyArrayNonEmpty(clientKeys) + || isReadonlyArrayNonEmpty(serverKeys) + || DataDependencies.isNonEmpty(writeDependencies) + ) { + successfulInvalidations.set(cacheKey, { clientKeys, serverKeys, writeDependencies }) + } else { + successfulInvalidations.delete(cacheKey) + } + return Effect.void + }) + const invalidateCache = ( input: unknown, output: Exit.Exit, @@ -318,7 +372,28 @@ const buildInvalidateCache = ( const derivedKeys = getDerivedInvalidationKeys(writeDependencies) // Invalidate exact reactivity keys (= the prefixes query atoms register under). Each key // array is hashed structurally, matching the query-side registration. - const keys: ReadonlyArray> = [...clientKeys, ...serverKeys, ...derivedKeys] + const cacheKey = invalidateOnFailureFromLastSuccess ? mutationKey(input) : undefined + const cached = cacheKey === undefined || output._tag === "Success" + ? undefined + : successfulInvalidations.get(cacheKey) + const cachedKeys = cached === undefined ? [] : [ + ...cached.clientKeys, + ...cached.serverKeys, + ...getDerivedInvalidationKeys(cached.writeDependencies) + ] + const keys = uniqueKeys([...clientKeys, ...serverKeys, ...derivedKeys, ...cachedKeys]) + + if (cacheKey !== undefined && output._tag === "Success") { + if ( + isReadonlyArrayNonEmpty(clientKeys) + || isReadonlyArrayNonEmpty(serverKeys) + || DataDependencies.isNonEmpty(writeDependencies) + ) { + successfulInvalidations.set(cacheKey, { clientKeys, serverKeys, writeDependencies }) + } else { + successfulInvalidations.delete(cacheKey) + } + } if (!isReadonlyArrayNonEmpty(keys)) return Effect.void @@ -341,7 +416,7 @@ const buildInvalidateCache = ( ) }) - return invalidateCache + return Object.assign(invalidateCache, { rememberSuccess }) } export const invalidateQueries = ( @@ -349,7 +424,12 @@ export const invalidateQueries = ( options: MutationOptionsBase | undefined, queryInvalidator: QueryInvalidator ) => { - const invalidateCache = buildInvalidateCache(self, options?.queryInvalidation, queryInvalidator) + const invalidateCache = buildInvalidateCache( + self, + options?.queryInvalidation, + options?.invalidateOnFailureFromLastSuccess, + queryInvalidator + ) const select = options?.select @@ -448,16 +528,30 @@ export const makeStreamMutation2 = (queryInvalidator: QueryInvalid }, mergedInvalidation?: MutationOptionsBase["queryInvalidation"] ) => { - const invCache = buildInvalidateCache(self, mergedInvalidation, queryInvalidator) + const liveInvCache = buildInvalidateCache(self, mergedInvalidation, undefined, queryInvalidator) - const makeInvocationEffect = (input: unknown, source: Stream.Stream) => + const makeInvocationEffect = ( + input: unknown, + source: Stream.Stream, + options?: Pick + ) => Effect.gen(function*() { + const invCache = buildInvalidateCache( + self, + mergedInvalidation, + options?.invalidateOnFailureFromLastSuccess, + queryInvalidator + ) const keysRef = yield* Ref.make>([]) + const handledKeysRef = yield* Ref.make>([]) const invKeys = makeInvalidationKeysService( keysRef, // Stream invalidation is sequenced by the injected query invalidator; this callback // returns void to keep the server-side invalidation service effect-free. - (key) => invCache(input, Exit.succeed(undefined), [key]) as Effect.Effect + (key) => + Ref + .update(handledKeysRef, (keys) => [...keys, key]) + .pipe(Effect.andThen(liveInvCache(input, Exit.succeed(undefined), [key]))) as Effect.Effect ) const readsRef = yield* Ref.make(DataDependencies.empty()) const writesRef = yield* Ref.make(DataDependencies.empty()) @@ -467,17 +561,28 @@ export const makeStreamMutation2 = (queryInvalidator: QueryInvalid Stream.provideService(InvalidationKeysFromServer, invKeys), Stream.provideService(DataDependencies.DataDependencyRecorder, dependencyRecorder), Stream.tap((v) => Ref.set(lastRef, v)), - Stream.ensuring( + Stream.onExit((exit) => Effect.gen(function*() { const lastValue = yield* Ref.get(lastRef) const serverKeys = yield* Ref.get(keysRef) + const handledServerKeys = yield* Ref.get(handledKeysRef) const writeDependencies = yield* Ref.get(writesRef) - yield* invCache(input, Exit.succeed(lastValue), serverKeys, writeDependencies) + const output = exit._tag === "Success" ? Exit.succeed(lastValue) : exit + yield* invCache(input, output, serverKeys, writeDependencies) + if (exit._tag === "Success" && isReadonlyArrayNonEmpty(handledServerKeys)) { + yield* invCache.rememberSuccess( + input, + output, + [...handledServerKeys, ...serverKeys], + writeDependencies + ) + } }) ) ) }) - return (i: any) => Stream.unwrap(makeInvocationEffect(i, self.handler(i))) + return (i: any, options?: Pick) => + Stream.unwrap(makeInvocationEffect(i, self.handler(i), options)) } } diff --git a/packages/vue/test/dependencyInvalidation.test.ts b/packages/vue/test/dependencyInvalidation.test.ts index 9330ba304..3a28f2143 100644 --- a/packages/vue/test/dependencyInvalidation.test.ts +++ b/packages/vue/test/dependencyInvalidation.test.ts @@ -5,16 +5,18 @@ import { QueryClient, VueQueryPlugin } from "@tanstack/vue-query" import { DataDependencies, type InvalidationKey, InvalidationKeysFromServer, makeQueryKey } from "effect-app/client" import * as Context from "effect-app/Context" import * as Effect from "effect-app/Effect" +import * as Exit from "effect/Exit" import * as Fiber from "effect/Fiber" import * as Layer from "effect/Layer" import * as ManagedRuntime from "effect/ManagedRuntime" +import * as Stream from "effect/Stream" import { TestClock } from "effect/testing" import * as Reactivity from "effect/unstable/reactivity/Reactivity" import { createApp, effectScope, ref } from "vue" import { awaitAtomResult, buildQueryFamily, invalidateAndAwait, makeAtomClientRuntime } from "../src/atomQuery.js" import { clearQueryReadDependencies, getDerivedInvalidationKeys, setQueryReadDependencies } from "../src/dependencyMetadata.js" import { makeTanstackQuery, makeTanstackQueryInvalidator } from "../src/internal/tanstackQuery.js" -import { invalidateQueries, type MutationOptionsBase } from "../src/mutate.js" +import { invalidateQueries, makeStreamMutation2, type MutationOptionsBase } from "../src/mutate.js" const repo = DataDependencies.repo("FrontendRepo") const otherRepo = DataDependencies.repo("OtherRepo") @@ -93,6 +95,97 @@ it.effect("a command's write deps invalidate active queries whose recorded reads expect(recorded).toContainEqual(inventoryKey) })) +it.effect("opt-in cached success invalidation is replayed when the same mutation input fails", () => + Effect.gen(function*() { + const inventoryKey = ["$CachedSuccess", "List", undefined] + const cachedRepo = DataDependencies.repo("CachedSuccessRepo") + const recorded: Array> = [] + const queryInvalidator = { + invalidateAndAwait: (keys: ReadonlyArray>) => + Effect.sync(() => keys.forEach((key) => recorded.push(key))) + } + const mutate = invalidateQueries( + { id: "CachedSuccess.Save" }, + { invalidateOnFailureFromLastSuccess: true }, + queryInvalidator + ) + + yield* mutate(DataDependencies.write(cachedRepo).pipe(Effect.as(123)), { id: "abc" }) + expect(recorded).toStrictEqual([]) + + setQueryReadDependencies(inventoryKey, new Set([cachedRepo])) + try { + const fiber = yield* Effect.forkChild(mutate(Effect.fail("boom"), { id: "abc" }).pipe(Effect.exit)) + yield* TestClock.adjust("1 millis") + const exit = yield* Fiber.join(fiber) + + expect(Exit.isFailure(exit)).toBe(true) + expect(recorded).toContainEqual(inventoryKey) + } finally { + clearQueryReadDependencies(inventoryKey) + } + })) + +it.effect("cached success invalidation is opt-in", () => + Effect.gen(function*() { + const inventoryKey = ["$CachedSuccessOptOut", "List", undefined] + const cachedRepo = DataDependencies.repo("CachedSuccessOptOutRepo") + const recorded: Array> = [] + const queryInvalidator = { + invalidateAndAwait: (keys: ReadonlyArray>) => + Effect.sync(() => keys.forEach((key) => recorded.push(key))) + } + const mutate = invalidateQueries({ id: "CachedSuccessOptOut.Save" }, undefined, queryInvalidator) + + yield* mutate(DataDependencies.write(cachedRepo).pipe(Effect.as(123)), { id: "abc" }) + setQueryReadDependencies(inventoryKey, new Set([cachedRepo])) + try { + const exit = yield* mutate(Effect.fail("boom"), { id: "abc" }).pipe(Effect.exit) + + expect(Exit.isFailure(exit)).toBe(true) + expect(recorded).toStrictEqual([]) + } finally { + clearQueryReadDependencies(inventoryKey) + } + })) + +it.effect("stream mutations can replay cached success invalidation on failure", () => + Effect.gen(function*() { + const inventoryKey = ["$CachedStreamSuccess", "List", undefined] + const cachedRepo = DataDependencies.repo("CachedStreamSuccessRepo") + const recorded: Array> = [] + const queryInvalidator = { + invalidateAndAwait: (keys: ReadonlyArray>) => + Effect.sync(() => keys.forEach((key) => recorded.push(key))) + } + let shouldFail = false + const mutate = makeStreamMutation2(queryInvalidator)({ + id: "CachedStreamSuccess.Save", + handler: () => + shouldFail + ? Stream.fail("boom") + : Stream.fromEffect(DataDependencies.write(cachedRepo).pipe(Effect.as(1))) + }) + + yield* Stream.runDrain(mutate({ id: "abc" }, { invalidateOnFailureFromLastSuccess: true })) + expect(recorded).toStrictEqual([]) + + shouldFail = true + setQueryReadDependencies(inventoryKey, new Set([cachedRepo])) + try { + const fiber = yield* Effect.forkChild( + Stream.runDrain(mutate({ id: "abc" }, { invalidateOnFailureFromLastSuccess: true })).pipe(Effect.exit) + ) + yield* TestClock.adjust("1 millis") + const exit = yield* Fiber.join(fiber) + + expect(Exit.isFailure(exit)).toBe(true) + expect(recorded).toContainEqual(inventoryKey) + } finally { + clearQueryReadDependencies(inventoryKey) + } + })) + // --- atom engine: recording wires through buildQueryFamily --------------------------------------- it("atom engine: a query records its read deps so a command's writes derive it", async () => {