fix: retain uncertain optimistic mutations

2026-08-02 02:42:56 +09:00
parent 15645541b7
commit d9afccdd60
15 changed files with 2172 additions and 124 deletions
@@ -3,8 +3,11 @@ import {
  useEffect,
  useMemo,
  useState,
+  useSyncExternalStore,
 } from "react";
 import {
+  hashKey,
+  type QueryClient,
  useMutation,
  useQuery,
  useQueryClient,
@@ -18,14 +21,20 @@ import type { Result } from "../../../application/result.ts";
 import {
  createFailure,
  normalizeUnknownFailure,
+  withFailureEffect,
  type AppFailure,
 } from "../../../contracts/errors.ts";
-import type { QueryInvalidationTopic } from "../../../contracts/query-invalidation.ts";
+import type {
+  QueryInvalidationCoordinator,
+  QueryInvalidationTopic,
+} from "../../../contracts/query-invalidation.ts";
 import type { MutationIntent } from "../../../contracts/mutation-intent.ts";
+import type { CacheScopeSnapshot } from "../../../contracts/server-state-scope.ts";
 import {
  admitQueryResult,
  type BoundMutation,
  type BoundQuery,
+  MUTATION_COORDINATOR_BOUNDS,
  type MutationDuplicatePolicy,
 } from "../../../contracts/server-state.ts";
 import { runtimeIdentityToken } from "../../../contracts/query-keys.ts";
@@ -181,27 +190,75 @@ export function useApplicationQuery<Value>(
  });
 }

-type LegacyMutationOptions<Input, Value> = Readonly<{
+type LegacyMutationBase<Input, Value> = Readonly<{
  execute(input: Input): Promise<ApplicationResult<Value>>;
  duplicatePolicy?: MutationDuplicatePolicy;
  invalidate?: readonly QueryInvalidationTopic[];
-  optimistic?: Readonly<{
-    queryKey: readonly unknown[];
-    update(previous: unknown, input: Input): unknown;
-  }>;
  currentData?: unknown;
 }>;

+type LegacyMutationOptions<Input, Value> = LegacyMutationBase<Input, Value> &
+  (
+    | Readonly<{
+        /** Stable logical identity required to retain an unknown effect across remounts. */
+        definitionId: string;
+        optimistic?: never;
+      }>
+    | Readonly<{
+        definitionId?: string;
+        optimistic: Readonly<{
+          queryKey: readonly unknown[];
+          update(previous: unknown, input: Input): unknown;
+        }>;
+      }>
+  );
+
 type ApplicationMutationController<Input, Value> = Readonly<{
  state: AsyncState;
  submit(input: Input): Promise<ApplicationResult<Value>>;
  resolveConflict(): Promise<void>;
+  reconcileUnknownEffect(
+    resolution: "APPLIED" | "NOT_APPLIED",
+  ): Promise<void>;
 }>;

 type MutationExecution<Input> =
  | Readonly<{ kind: "BOUND"; input: Input; intent: MutationIntent }>
  | Readonly<{ kind: "LEGACY"; input: Input }>;

+type MutationAdmission = {
+  sequence: number;
+  state: "ACTIVE" | "UNKNOWN" | "RECONCILING" | "SETTLED";
+  intent: MutationIntent | null;
+  scope: CacheScopeSnapshot | null;
+  optimisticLayer: OptimisticLayerLease | null;
+  optimisticQueryKey: readonly unknown[] | null;
+  readonly invalidate: readonly QueryInvalidationTopic[];
+  invalidationCoordinator: QueryInvalidationCoordinator | null;
+};
+
+type UnknownEffectChannel = {
+  readonly key: string;
+  readonly registry: UnknownEffectRegistry;
+  readonly owner: QueryClient;
+  readonly scope: CacheScopeSnapshot | null;
+  nextSequence: number;
+  admissions: MutationAdmission[];
+  version: number;
+  reconciliationInFlight: boolean;
+  readonly listeners: Set<() => void>;
+  disposeScopeListener: (() => void) | null;
+};
+
+type UnknownEffectRegistry = {
+  readonly channels: Map<string, UnknownEffectChannel>;
+  activeAdmissions: number;
+};
+
+const LEGACY_OPTIMISTIC_SCOPE = Object.freeze({
+  isCurrent: () => true,
+});
+
 export function useApplicationMutation<Input, Value>(
  options: BoundMutation<Input, Value>,
 ): ApplicationMutationController<Input, Value>;
@@ -221,7 +278,11 @@ export function useApplicationMutation<Input, Value>(
  const optimistic = "optimistic" in options ? options.optimistic : undefined;
  const currentData = "currentData" in options ? options.currentData : undefined;
  const definitionId =
-    "definitionId" in options ? options.definitionId : "LEGACY_MUTATION";
+    "definitionId" in options
+      ? options.definitionId ?? "LEGACY_MUTATION"
+      : "LEGACY_MUTATION";
+  const definitionVersion =
+    "definitionVersion" in options ? options.definitionVersion : 0;
  const duplicatePolicy =
    "duplicatePolicy" in options ? options.duplicatePolicy : "REJECT_WHILE_ACTIVE";
  const [conflict, setConflict] = useState<AppFailure | null>(null);
@@ -232,6 +293,53 @@ export function useApplicationMutation<Input, Value>(
    "requiresIdempotencyKey" in options
      ? options.requiresIdempotencyKey
      : false;
+  const unknownEffectChannelKey = scope
+    ? `bound:${scope.generation}:${scope.fingerprint}:${definitionId}:${definitionVersion}`
+    : `legacy:${definitionId}:${
+        optimistic
+          ? hashKey(optimistic.queryKey)
+          : "non-optimistic"
+      }`;
+  const [acquiredUnknownEffectChannel, setAcquiredUnknownEffectChannel] =
+    useState<UnknownEffectChannel | null>(null);
+  const unknownEffectChannel =
+    acquiredUnknownEffectChannel?.owner === queryClient &&
+    acquiredUnknownEffectChannel.key === unknownEffectChannelKey &&
+    acquiredUnknownEffectChannel.scope === (scope ?? null)
+      ? acquiredUnknownEffectChannel
+      : null;
+  useEffect(() => {
+    const acquired = acquireUnknownEffectChannel(
+      queryClient,
+      unknownEffectChannelKey,
+      scope,
+    );
+    setAcquiredUnknownEffectChannel(acquired);
+    return () => {
+      if (acquired) {
+        releaseUnknownEffectChannelIfUnused(queryClient, acquired);
+      }
+    };
+  }, [queryClient, scope, unknownEffectChannelKey]);
+  const subscribeToUnknownEffects = useCallback(
+    (listener: () => void) => {
+      if (!unknownEffectChannel) return () => {};
+      unknownEffectChannel.listeners.add(listener);
+      return () => {
+        unknownEffectChannel.listeners.delete(listener);
+        releaseUnknownEffectChannelIfUnused(
+          queryClient,
+          unknownEffectChannel,
+        );
+      };
+    },
+    [queryClient, unknownEffectChannel],
+  );
+  useSyncExternalStore(
+    subscribeToUnknownEffects,
+    () => unknownEffectChannel?.version ?? 0,
+    () => unknownEffectChannel?.version ?? 0,
+  );
  const mutation = useMutation<
    Value,
    ApplicationQueryError,
@@ -246,7 +354,7 @@ export function useApplicationMutation<Input, Value>(
            "SCOPE_GENERATION_CHANGED",
            definitionId,
            0,
-            { code: "MUTATION_SCOPE_STALE" },
+            { code: "MUTATION_SCOPE_STALE", effect: "NOT_STARTED" },
          ),
        );
      }
@@ -298,7 +406,7 @@ export function useApplicationMutation<Input, Value>(
                "SCOPE_GENERATION_CHANGED",
                definitionId,
                0,
-                { code: "MUTATION_SCOPE_STALE" },
+                { code: "MUTATION_SCOPE_STALE", effect: "NOT_STARTED" },
              ),
            });
          }
@@ -311,9 +419,12 @@ export function useApplicationMutation<Input, Value>(
      } catch (error) {
        return Promise.resolve({
          ok: false,
-          error: normalizeUnknownFailure(error, {
-            operationId: definitionId,
-          }),
+          error: withFailureEffect(
+            normalizeUnknownFailure(error, {
+              operationId: definitionId,
+            }),
+            "NOT_STARTED",
+          ),
        });
      }
      const active = mutationExecutions(queryClient).get(identity) as
@@ -331,26 +442,58 @@ export function useApplicationMutation<Input, Value>(
            "DUPLICATE_IN_FLIGHT",
            definitionId,
            0,
-            { code: "DUPLICATE_IN_FLIGHT" },
+            { code: "DUPLICATE_IN_FLIGHT", effect: "NOT_STARTED" },
          ),
        });
      }
+      if (
+        !unknownEffectChannel ||
+        unknownEffectChannel.registry.activeAdmissions >=
+          MUTATION_COORDINATOR_BOUNDS.activeIntentsTotal
+      ) {
+        identityLease?.release();
+        return Promise.resolve({
+          ok: false,
+          error: createFailure(
+            "IDENTITY_INTERN_LIMIT_EXCEEDED",
+            definitionId,
+            0,
+            {
+              code: "UNKNOWN_EFFECT_CHANNEL_LIMIT_EXCEEDED",
+              effect: "NOT_STARTED",
+            },
+          ),
+        });
+      }
+      const admission: MutationAdmission = {
+        sequence: unknownEffectChannel.nextSequence++,
+        state: "ACTIVE",
+        intent: null,
+        scope: scope ?? null,
+        optimisticLayer: null,
+        optimisticQueryKey: optimistic?.queryKey ?? null,
+        invalidate,
+        invalidationCoordinator: invalidationCoordinator ?? null,
+      };
+      unknownEffectChannel.admissions.push(admission);
+      unknownEffectChannel.registry.activeAdmissions += 1;
+      notifyUnknownEffectChannel(unknownEffectChannel);
      setConflict(null);
      mutation.reset();

      const pending = (async (): Promise<ApplicationResult<Value>> => {
-        const execution: MutationExecution<Input> =
-          scope
-            ? Object.freeze({
-                kind: "BOUND" as const,
-                input,
-                intent: mutationIntentFactory.create({
-                  operationId: mutationOperationId,
-                  canonicalInputIdentity: identity,
-                  requiresIdempotencyKey,
-                }),
-              })
-            : Object.freeze({ kind: "LEGACY" as const, input });
+        let execution: MutationExecution<Input>;
+        if (scope) {
+          const intent = mutationIntentFactory.create({
+            operationId: mutationOperationId,
+            canonicalInputIdentity: identity,
+            requiresIdempotencyKey,
+          });
+          admission.intent = intent;
+          execution = Object.freeze({ kind: "BOUND", input, intent });
+        } else {
+          execution = Object.freeze({ kind: "LEGACY", input });
+        }
        const mutationLease =
          invalidate.length === 0
            ? null
@@ -359,54 +502,60 @@ export function useApplicationMutation<Input, Value>(
          throw new Error("Query invalidation coordinator is not installed.");
        }
        try {
-          let previous: unknown;
-          let hadPreviousData = false;
          let optimisticLayer: OptimisticLayerLease | null = null;
          if (optimistic) {
            await queryClient.cancelQueries({
              queryKey: optimistic.queryKey,
              exact: true,
            });
-            if (scope) {
-              optimisticLayer = optimisticLayers(queryClient).begin(
-                optimistic.queryKey,
-                input,
-                optimistic.update,
-                scope,
-              );
-            } else {
-              previous = queryClient.getQueryData(optimistic.queryKey);
-              hadPreviousData = previous !== undefined;
-              queryClient.setQueryData(
-                optimistic.queryKey,
-                optimistic.update(previous, input),
-              );
-            }
+            optimisticLayer = optimisticLayers(queryClient).begin(
+              optimistic.queryKey,
+              input,
+              optimistic.update,
+              scope ?? LEGACY_OPTIMISTIC_SCOPE,
+            );
+            admission.optimisticLayer = optimisticLayer;
          }

          let value: Value;
          try {
            value = await mutation.mutateAsync(execution);
          } catch (error: unknown) {
-            if (optimistic) {
-              if (optimisticLayer) {
-                optimisticLayer.rollback();
-              } else if (hadPreviousData) {
-                queryClient.setQueryData(optimistic.queryKey, previous);
-              } else {
-                queryClient.removeQueries({
-                  queryKey: optimistic.queryKey,
-                  exact: true,
-                });
-              }
-            }
-            const failure =
+            const reportedFailure =
              error instanceof ApplicationQueryError
                ? error.failure
                : normalizeUnknownFailure(error, {
                    operationId: "APPLICATION_MUTATION",
                  });
-            if (failure.kind === "CONFLICT") setConflict(failure);
+            const failure = mutationFailureWithEffect(reportedFailure);
+            switch (failure.effect) {
+              case "NOT_STARTED":
+              case "NOT_APPLIED":
+                optimisticLayer?.rollback();
+                if (failure.kind === "CONFLICT") setConflict(failure);
+                settleMutationAdmission(unknownEffectChannel, admission);
+                break;
+              case "APPLIED_CONFIRMED":
+                optimisticLayer?.commit();
+                if (!scope || scope.isCurrent()) {
+                  try {
+                    await invalidationCoordinator?.invalidate(invalidate);
+                  } catch {
+                    // A confirmed command remains committed if refresh fails.
+                  }
+                }
+                settleMutationAdmission(unknownEffectChannel, admission);
+                break;
+              case "MAYBE_APPLIED":
+                optimisticLayer?.markUncertain();
+                if (!scope || scope.isCurrent()) {
+                  admission.state = "UNKNOWN";
+                  notifyUnknownEffectChannel(unknownEffectChannel);
+                } else {
+                  settleMutationAdmission(unknownEffectChannel, admission);
+                }
+                break;
+            }
            return { ok: false, error: failure };
          }

@@ -416,6 +565,7 @@ export function useApplicationMutation<Input, Value>(
          } catch {
            // Cache refresh remains best effort after the server has committed.
          }
+          settleMutationAdmission(unknownEffectChannel, admission);
          return { ok: true, value };
        } finally {
          try {
@@ -426,9 +576,13 @@ export function useApplicationMutation<Input, Value>(
        }
      })()
        .catch((error: unknown) => {
-          const failure = normalizeUnknownFailure(error, {
-            operationId: "APPLICATION_MUTATION",
-          });
+          const failure = withFailureEffect(
+            normalizeUnknownFailure(error, {
+              operationId: "APPLICATION_MUTATION",
+            }),
+            "NOT_STARTED",
+          );
+          settleMutationAdmission(unknownEffectChannel, admission);
          if (failure.kind === "CONFLICT") setConflict(failure);
          return { ok: false as const, error: failure };
        })
@@ -455,6 +609,7 @@ export function useApplicationMutation<Input, Value>(
      mutationOperationId,
      requiresIdempotencyKey,
      scope,
+      unknownEffectChannel,
    ],
  );

@@ -467,14 +622,194 @@ export function useApplicationMutation<Input, Value>(
    await invalidationCoordinator?.invalidate(invalidate);
  }, [invalidate, invalidationCoordinator, mutation]);

+  const reconcileUnknownEffect = useCallback(
+    async (resolution: "APPLIED" | "NOT_APPLIED") => {
+      if (
+        !unknownEffectChannel ||
+        unknownEffectChannel.reconciliationInFlight
+      ) {
+        return;
+      }
+      const pending = unknownEffectChannel.admissions[0];
+      if (!pending || pending.state !== "UNKNOWN") return;
+      if (pending.scope && !pending.intent) return;
+      unknownEffectChannel.reconciliationInFlight = true;
+      pending.state = "RECONCILING";
+      notifyUnknownEffectChannel(unknownEffectChannel);
+      try {
+        pending.optimisticLayer?.reconcile(resolution);
+        if (
+          resolution === "APPLIED" &&
+          (!pending.scope || pending.scope.isCurrent())
+        ) {
+          try {
+            await pending.invalidationCoordinator?.invalidate(
+              pending.invalidate,
+            );
+          } catch {
+            // Explicit reconciliation remains settled if refresh fails.
+          }
+        }
+      } finally {
+        if (pending.scope && !pending.scope.isCurrent()) {
+          removeAdmissionOptimisticQuery(queryClient, pending);
+        }
+        settleMutationAdmission(unknownEffectChannel, pending);
+        if (
+          !unknownEffectChannel.admissions.some(
+            (admission) => admission.state === "ACTIVE",
+          )
+        ) {
+          mutation.reset();
+        }
+        // Keep the head intent locked through the current event turn so two
+        // clicks cannot consume two FIFO records when reconciliation itself
+        // has no asynchronous invalidation work.
+        await Promise.resolve();
+        unknownEffectChannel.reconciliationInFlight = false;
+        notifyUnknownEffectChannel(unknownEffectChannel);
+      }
+    },
+    [mutation, queryClient, unknownEffectChannel],
+  );
+
+  const unknownEffectHead = unknownEffectChannel?.admissions[0];
+
  return Object.freeze({
    state: deriveAsyncState({
      data: currentData ?? true,
-      isMutationPending: mutation.isPending,
+      isMutationPending:
+        mutation.isPending ||
+        unknownEffectHead?.state === "ACTIVE" ||
+        unknownEffectHead?.state === "RECONCILING",
+      hasMutationEffectUnknown: unknownEffectHead?.state === "UNKNOWN",
      hasMutationConflict: conflict !== null,
    }),
    submit,
    resolveConflict,
+    reconcileUnknownEffect,
+  });
+}
+
+function mutationFailureWithEffect(failure: AppFailure): AppFailure {
+  switch (failure.effect) {
+    case "NOT_STARTED":
+    case "NOT_APPLIED":
+    case "APPLIED_CONFIRMED":
+    case "MAYBE_APPLIED":
+      return withFailureEffect(failure, failure.effect);
+    case "NOT_APPLICABLE":
+    case undefined:
+      return withFailureEffect(failure, "MAYBE_APPLIED");
+  }
+}
+
+const UNKNOWN_EFFECT_CHANNELS = new WeakMap<
+  object,
+  UnknownEffectRegistry
+>();
+
+function acquireUnknownEffectChannel(
+  queryClient: QueryClient,
+  key: string,
+  scope: CacheScopeSnapshot | undefined,
+): UnknownEffectChannel | null {
+  let registry = UNKNOWN_EFFECT_CHANNELS.get(queryClient);
+  if (!registry) {
+    registry = { channels: new Map(), activeAdmissions: 0 };
+    UNKNOWN_EFFECT_CHANNELS.set(queryClient, registry);
+  }
+  const { channels } = registry;
+  const existing = channels.get(key);
+  if (existing) return existing.scope === (scope ?? null) ? existing : null;
+  if (
+    channels.size >= MUTATION_COORDINATOR_BOUNDS.activeDefinitionsPerRuntime
+  ) {
+    return null;
+  }
+  const created: UnknownEffectChannel = {
+    key,
+    registry,
+    owner: queryClient,
+    scope: scope ?? null,
+    nextSequence: 1,
+    admissions: [],
+    version: 0,
+    reconciliationInFlight: false,
+    listeners: new Set(),
+    disposeScopeListener: null,
+  };
+  channels.set(key, created);
+  if (scope) {
+    const discardScope = () => {
+      for (const admission of created.admissions) {
+        removeAdmissionOptimisticQuery(queryClient, admission);
+        markMutationAdmissionSettled(created, admission);
+      }
+      created.admissions = [];
+      notifyUnknownEffectChannel(created);
+      releaseUnknownEffectChannelIfUnused(queryClient, created);
+    };
+    scope.signal.addEventListener("abort", discardScope, { once: true });
+    created.disposeScopeListener = () =>
+      scope.signal.removeEventListener("abort", discardScope);
+    if (scope.signal.aborted || !scope.isCurrent()) discardScope();
+  }
+  return created;
+}
+
+function releaseUnknownEffectChannelIfUnused(
+  queryClient: QueryClient,
+  channel: UnknownEffectChannel,
+): void {
+  if (channel.admissions.length > 0 || channel.listeners.size > 0) return;
+  const registry = UNKNOWN_EFFECT_CHANNELS.get(queryClient);
+  const channels = registry?.channels;
+  if (channels?.get(channel.key) !== channel) return;
+  channel.disposeScopeListener?.();
+  channels.delete(channel.key);
+  if (channels.size === 0 && registry?.activeAdmissions === 0) {
+    UNKNOWN_EFFECT_CHANNELS.delete(queryClient);
+  }
+}
+
+function notifyUnknownEffectChannel(channel: UnknownEffectChannel): void {
+  channel.version += 1;
+  for (const listener of [...channel.listeners]) listener();
+}
+
+function settleMutationAdmission(
+  channel: UnknownEffectChannel,
+  admission: MutationAdmission,
+): void {
+  markMutationAdmissionSettled(channel, admission);
+  while (channel.admissions[0]?.state === "SETTLED") {
+    channel.admissions.shift();
+  }
+  notifyUnknownEffectChannel(channel);
+  releaseUnknownEffectChannelIfUnused(channel.owner, channel);
+}
+
+function markMutationAdmissionSettled(
+  channel: UnknownEffectChannel,
+  admission: MutationAdmission,
+): void {
+  if (admission.state === "SETTLED") return;
+  admission.state = "SETTLED";
+  channel.registry.activeAdmissions = Math.max(
+    0,
+    channel.registry.activeAdmissions - 1,
+  );
+}
+
+function removeAdmissionOptimisticQuery(
+  queryClient: QueryClient,
+  admission: MutationAdmission,
+): void {
+  if (!admission.optimisticQueryKey) return;
+  queryClient.removeQueries({
+    queryKey: admission.optimisticQueryKey,
+    exact: true,
  });
 }