feat(runtime-core, reactivity): `onEffectCleanup` and `baseWatch` (#82)

Co-authored-by: 三咲智子 Kevin Deng <sxzz@sxzz.moe>

packages/reactivity/__tests__/baseWatch.spec.ts

@@ -0,0 +1,178 @@
+import type { Scheduler, SchedulerJob } from '../src/baseWatch'
+import {
+  BaseWatchErrorCodes,
+  EffectScope,
+  type Ref,
+  baseWatch,
+  onEffectCleanup,
+  ref,
+} from '../src'
+
+const queue: SchedulerJob[] = []
+
+// these codes are a simple scheduler
+let isFlushPending = false
+const resolvedPromise = /*#__PURE__*/ Promise.resolve() as Promise<any>
+const nextTick = (fn?: () => any) =>
+  fn ? resolvedPromise.then(fn) : resolvedPromise
+const scheduler: Scheduler = job => {
+  queue.push(job)
+  flushJobs()
+}
+const flushJobs = () => {
+  if (isFlushPending) return
+  isFlushPending = true
+  resolvedPromise.then(() => {
+    queue.forEach(job => job())
+    queue.length = 0
+    isFlushPending = false
+  })
+}
+
+describe('baseWatch', () => {
+  test('effect', () => {
+    let dummy: any
+    const source = ref(0)
+    baseWatch(() => {
+      dummy = source.value
+    })
+    expect(dummy).toBe(0)
+    source.value++
+    expect(dummy).toBe(1)
+  })
+
+  test('watch', () => {
+    let dummy: any
+    const source = ref(0)
+    baseWatch(source, () => {
+      dummy = source.value
+    })
+    expect(dummy).toBe(undefined)
+    source.value++
+    expect(dummy).toBe(1)
+  })
+
+  test('custom error handler', () => {
+    const onError = vi.fn()
+
+    baseWatch(
+      () => {
+        throw 'oops in effect'
+      },
+      null,
+      { onError },
+    )
+
+    const source = ref(0)
+    const effect = baseWatch(
+      source,
+      () => {
+        onEffectCleanup(() => {
+          throw 'oops in cleanup'
+        })
+        throw 'oops in watch'
+      },
+      { onError },
+    )
+
+    expect(onError.mock.calls.length).toBe(1)
+    expect(onError.mock.calls[0]).toMatchObject([
+      'oops in effect',
+      BaseWatchErrorCodes.WATCH_CALLBACK,
+    ])
+
+    source.value++
+    expect(onError.mock.calls.length).toBe(2)
+    expect(onError.mock.calls[1]).toMatchObject([
+      'oops in watch',
+      BaseWatchErrorCodes.WATCH_CALLBACK,
+    ])
+
+    effect!.stop()
+    source.value++
+    expect(onError.mock.calls.length).toBe(3)
+    expect(onError.mock.calls[2]).toMatchObject([
+      'oops in cleanup',
+      BaseWatchErrorCodes.WATCH_CLEANUP,
+    ])
+  })
+
+  test('baseWatch with onEffectCleanup', async () => {
+    let dummy = 0
+    let source: Ref<number>
+    const scope = new EffectScope()
+
+    scope.run(() => {
+      source = ref(0)
+      baseWatch(onCleanup => {
+        source.value
+
+        onCleanup(() => (dummy += 2))
+        onEffectCleanup(() => (dummy += 3))
+        onEffectCleanup(() => (dummy += 5))
+      })
+    })
+    expect(dummy).toBe(0)
+
+    scope.run(() => {
+      source.value++
+    })
+    expect(dummy).toBe(10)
+
+    scope.run(() => {
+      source.value++
+    })
+    expect(dummy).toBe(20)
+
+    scope.stop()
+    expect(dummy).toBe(30)
+  })
+
+  test('nested calls to baseWatch and onEffectCleanup', async () => {
+    let calls: string[] = []
+    let source: Ref<number>
+    let copyist: Ref<number>
+    const scope = new EffectScope()
+
+    scope.run(() => {
+      source = ref(0)
+      copyist = ref(0)
+      // sync by default
+      baseWatch(
+        () => {
+          const current = (copyist.value = source.value)
+          onEffectCleanup(() => calls.push(`sync ${current}`))
+        },
+        null,
+        {},
+      )
+      // with scheduler
+      baseWatch(
+        () => {
+          const current = copyist.value
+          onEffectCleanup(() => calls.push(`post ${current}`))
+        },
+        null,
+        { scheduler },
+      )
+    })
+
+    await nextTick()
+    expect(calls).toEqual([])
+
+    scope.run(() => source.value++)
+    expect(calls).toEqual(['sync 0'])
+    await nextTick()
+    expect(calls).toEqual(['sync 0', 'post 0'])
+    calls.length = 0
+
+    scope.run(() => source.value++)
+    expect(calls).toEqual(['sync 1'])
+    await nextTick()
+    expect(calls).toEqual(['sync 1', 'post 1'])
+    calls.length = 0
+
+    scope.stop()
+    expect(calls).toEqual(['sync 2', 'post 2'])
+  })
+})

packages/reactivity/src/baseWatch.ts

@@ -0,0 +1,417 @@
+import {
+  EMPTY_OBJ,
+  NOOP,
+  hasChanged,
+  isArray,
+  isFunction,
+  isMap,
+  isObject,
+  isPlainObject,
+  isPromise,
+  isSet,
+} from '@vue/shared'
+import { warn } from './warning'
+import type { ComputedRef } from './computed'
+import { ReactiveFlags } from './constants'
+import {
+  type DebuggerOptions,
+  type EffectScheduler,
+  ReactiveEffect,
+  pauseTracking,
+  resetTracking,
+} from './effect'
+import { isReactive, isShallow } from './reactive'
+import { type Ref, isRef } from './ref'
+import { getCurrentScope } from './effectScope'
+
+// These errors were transferred from `packages/runtime-core/src/errorHandling.ts`
+// along with baseWatch to maintain code compatibility. Hence,
+// it is essential to keep these values unchanged.
+export enum BaseWatchErrorCodes {
+  WATCH_GETTER = 2,
+  WATCH_CALLBACK,
+  WATCH_CLEANUP,
+}
+
+// TODO move to a scheduler package
+export interface SchedulerJob extends Function {
+  id?: number
+  pre?: boolean
+  active?: boolean
+  computed?: boolean
+  /**
+   * Indicates whether the effect is allowed to recursively trigger itself
+   * when managed by the scheduler.
+   *
+   * By default, a job cannot trigger itself because some built-in method calls,
+   * e.g. Array.prototype.push actually performs reads as well (#1740) which
+   * can lead to confusing infinite loops.
+   * The allowed cases are component update functions and watch callbacks.
+   * Component update functions may update child component props, which in turn
+   * trigger flush: "pre" watch callbacks that mutates state that the parent
+   * relies on (#1801). Watch callbacks doesn't track its dependencies so if it
+   * triggers itself again, it's likely intentional and it is the user's
+   * responsibility to perform recursive state mutation that eventually
+   * stabilizes (#1727).
+   */
+  allowRecurse?: boolean
+}
+
+type WatchEffect = (onCleanup: OnCleanup) => void
+type WatchSource<T = any> = Ref<T> | ComputedRef<T> | (() => T)
+type WatchCallback<V = any, OV = any> = (
+  value: V,
+  oldValue: OV,
+  onCleanup: OnCleanup,
+) => any
+type OnCleanup = (cleanupFn: () => void) => void
+
+export interface BaseWatchOptions<Immediate = boolean> extends DebuggerOptions {
+  immediate?: Immediate
+  deep?: boolean
+  once?: boolean
+  scheduler?: Scheduler
+  onError?: HandleError
+  onWarn?: HandleWarn
+}
+
+// initial value for watchers to trigger on undefined initial values
+const INITIAL_WATCHER_VALUE = {}
+
+export type Scheduler = (
+  job: SchedulerJob,
+  effect: ReactiveEffect,
+  isInit: boolean,
+) => void
+export type HandleError = (err: unknown, type: BaseWatchErrorCodes) => void
+export type HandleWarn = (msg: string, ...args: any[]) => void
+
+const DEFAULT_SCHEDULER: Scheduler = job => job()
+const DEFAULT_HANDLE_ERROR: HandleError = (err: unknown) => {
+  throw err
+}
+
+const cleanupMap: WeakMap<ReactiveEffect, (() => void)[]> = new WeakMap()
+let activeEffect: ReactiveEffect | undefined = undefined
+
+/**
+ * Returns the current active effect if there is one.
+ */
+export function getCurrentEffect() {
+  return activeEffect
+}
+
+/**
+ * Registers a cleanup callback on the current active effect. This
+ * registered cleanup callback will be invoked right before the
+ * associated effect re-runs.
+ *
+ * @param cleanupFn - The callback function to attach to the effect's cleanup.
+ */
+export function onEffectCleanup(cleanupFn: () => void) {
+  if (activeEffect) {
+    const cleanups =
+      cleanupMap.get(activeEffect) ||
+      cleanupMap.set(activeEffect, []).get(activeEffect)!
+    cleanups.push(cleanupFn)
+  } else if (__DEV__) {
+    warn(
+      `onEffectCleanup() was called when there was no active effect` +
+        ` to associate with.`,
+    )
+  }
+}
+
+export function baseWatch(
+  source: WatchSource | WatchSource[] | WatchEffect | object,
+  cb?: WatchCallback | null,
+  {
+    immediate,
+    deep,
+    once,
+    scheduler = DEFAULT_SCHEDULER,
+    onWarn = __DEV__ ? warn : NOOP,
+    onError = DEFAULT_HANDLE_ERROR,
+    onTrack,
+    onTrigger,
+  }: BaseWatchOptions = EMPTY_OBJ,
+): ReactiveEffect | undefined {
+  const warnInvalidSource = (s: unknown) => {
+    onWarn(
+      `Invalid watch source: `,
+      s,
+      `A watch source can only be a getter/effect function, a ref, ` +
+        `a reactive object, or an array of these types.`,
+    )
+  }
+
+  const reactiveGetter = (source: object) =>
+    deep === true
+      ? source // traverse will happen in wrapped getter below
+      : // for deep: false, only traverse root-level properties
+        traverse(source, deep === false ? 1 : undefined)
+
+  let effect: ReactiveEffect
+  let getter: () => any
+  let cleanup: (() => void) | undefined
+  let forceTrigger = false
+  let isMultiSource = false
+
+  if (isRef(source)) {
+    getter = () => source.value
+    forceTrigger = isShallow(source)
+  } else if (isReactive(source)) {
+    getter = () => reactiveGetter(source)
+    forceTrigger = true
+  } else if (isArray(source)) {
+    isMultiSource = true
+    forceTrigger = source.some(s => isReactive(s) || isShallow(s))
+    getter = () =>
+      source.map(s => {
+        if (isRef(s)) {
+          return s.value
+        } else if (isReactive(s)) {
+          return reactiveGetter(s)
+        } else if (isFunction(s)) {
+          return callWithErrorHandling(
+            s,
+            onError,
+            BaseWatchErrorCodes.WATCH_GETTER,
+          )
+        } else {
+          __DEV__ && warnInvalidSource(s)
+        }
+      })
+  } else if (isFunction(source)) {
+    if (cb) {
+      // getter with cb
+      getter = () =>
+        callWithErrorHandling(source, onError, BaseWatchErrorCodes.WATCH_GETTER)
+    } else {
+      // no cb -> simple effect
+      getter = () => {
+        if (cleanup) {
+          pauseTracking()
+          try {
+            cleanup()
+          } finally {
+            resetTracking()
+          }
+        }
+        const currentEffect = activeEffect
+        activeEffect = effect
+        try {
+          return callWithAsyncErrorHandling(
+            source,
+            onError,
+            BaseWatchErrorCodes.WATCH_CALLBACK,
+            [onEffectCleanup],
+          )
+        } finally {
+          activeEffect = currentEffect
+        }
+      }
+    }
+  } else {
+    getter = NOOP
+    __DEV__ && warnInvalidSource(source)
+  }
+
+  if (cb && deep) {
+    const baseGetter = getter
+    getter = () => traverse(baseGetter())
+  }
+
+  if (once) {
+    if (!cb) {
+      // onEffectCleanup need use effect as a key
+      getCurrentScope()?.effects.push((effect = {} as any))
+      getter()
+      return
+    }
+    if (immediate) {
+      // onEffectCleanup need use effect as a key
+      getCurrentScope()?.effects.push((effect = {} as any))
+      callWithAsyncErrorHandling(
+        cb,
+        onError,
+        BaseWatchErrorCodes.WATCH_CALLBACK,
+        [getter(), isMultiSource ? [] : undefined, onEffectCleanup],
+      )
+      return
+    }
+    const _cb = cb
+    cb = (...args) => {
+      _cb(...args)
+      effect?.stop()
+    }
+  }
+
+  let oldValue: any = isMultiSource
+    ? new Array((source as []).length).fill(INITIAL_WATCHER_VALUE)
+    : INITIAL_WATCHER_VALUE
+  const job: SchedulerJob = () => {
+    if (!effect.active || !effect.dirty) {
+      return
+    }
+    if (cb) {
+      // watch(source, cb)
+      const newValue = effect.run()
+      if (
+        deep ||
+        forceTrigger ||
+        (isMultiSource
+          ? (newValue as any[]).some((v, i) => hasChanged(v, oldValue[i]))
+          : hasChanged(newValue, oldValue))
+      ) {
+        // cleanup before running cb again
+        if (cleanup) {
+          cleanup()
+        }
+        const currentEffect = activeEffect
+        activeEffect = effect
+        try {
+          callWithAsyncErrorHandling(
+            cb,
+            onError,
+            BaseWatchErrorCodes.WATCH_CALLBACK,
+            [
+              newValue,
+              // pass undefined as the old value when it's changed for the first time
+              oldValue === INITIAL_WATCHER_VALUE
+                ? undefined
+                : isMultiSource && oldValue[0] === INITIAL_WATCHER_VALUE
+                  ? []
+                  : oldValue,
+              onEffectCleanup,
+            ],
+          )
+          oldValue = newValue
+        } finally {
+          activeEffect = currentEffect
+        }
+      }
+    } else {
+      // watchEffect
+      effect.run()
+    }
+  }
+
+  // important: mark the job as a watcher callback so that scheduler knows
+  // it is allowed to self-trigger (#1727)
+  job.allowRecurse = !!cb
+
+  let effectScheduler: EffectScheduler = () => scheduler(job, effect, false)
+
+  effect = new ReactiveEffect(getter, NOOP, effectScheduler)
+
+  cleanup = effect.onStop = () => {
+    const cleanups = cleanupMap.get(effect)
+    if (cleanups) {
+      cleanups.forEach(cleanup =>
+        callWithErrorHandling(
+          cleanup,
+          onError,
+          BaseWatchErrorCodes.WATCH_CLEANUP,
+        ),
+      )
+      cleanupMap.delete(effect)
+    }
+  }
+
+  if (__DEV__) {
+    effect.onTrack = onTrack
+    effect.onTrigger = onTrigger
+  }
+
+  // initial run
+  if (cb) {
+    if (immediate) {
+      job()
+    } else {
+      oldValue = effect.run()
+    }
+  } else {
+    scheduler(job, effect, true)
+  }
+
+  return effect
+}
+
+export function traverse(
+  value: unknown,
+  depth?: number,
+  currentDepth = 0,
+  seen?: Set<unknown>,
+) {
+  if (!isObject(value) || (value as any)[ReactiveFlags.SKIP]) {
+    return value
+  }
+
+  if (depth && depth > 0) {
+    if (currentDepth >= depth) {
+      return value
+    }
+    currentDepth++
+  }
+
+  seen = seen || new Set()
+  if (seen.has(value)) {
+    return value
+  }
+  seen.add(value)
+  if (isRef(value)) {
+    traverse(value.value, depth, currentDepth, seen)
+  } else if (isArray(value)) {
+    for (let i = 0; i < value.length; i++) {
+      traverse(value[i], depth, currentDepth, seen)
+    }
+  } else if (isSet(value) || isMap(value)) {
+    value.forEach((v: any) => {
+      traverse(v, depth, currentDepth, seen)
+    })
+  } else if (isPlainObject(value)) {
+    for (const key in value) {
+      traverse(value[key], depth, currentDepth, seen)
+    }
+  }
+  return value
+}
+
+function callWithErrorHandling(
+  fn: Function,
+  handleError: HandleError,
+  type: BaseWatchErrorCodes,
+  args?: unknown[],
+) {
+  let res
+  try {
+    res = args ? fn(...args) : fn()
+  } catch (err) {
+    handleError(err, type)
+  }
+  return res
+}
+
+function callWithAsyncErrorHandling(
+  fn: Function | Function[],
+  handleError: HandleError,
+  type: BaseWatchErrorCodes,
+  args?: unknown[],
+): any[] {
+  if (isFunction(fn)) {
+    const res = callWithErrorHandling(fn, handleError, type, args)
+    if (res && isPromise(res)) {
+      res.catch(err => {
+        handleError(err, type)
+      })
+    }
+    return res
+  }
+
+  const values = []
+  for (let i = 0; i < fn.length; i++) {
+    values.push(callWithAsyncErrorHandling(fn[i], handleError, type, args))
+  }
+  return values
+}

packages/reactivity/src/index.ts

@@ -69,3 +69,11 @@ export {
   onScopeDispose,
 } from './effectScope'
 export { TrackOpTypes, TriggerOpTypes, ReactiveFlags } from './constants'
+export {
+  baseWatch,
+  onEffectCleanup,
+  traverse,
+  BaseWatchErrorCodes,
+  type BaseWatchOptions,
+  type Scheduler,
+} from './baseWatch'

packages/runtime-core/__tests__/apiWatch.spec.ts

@@ -5,6 +5,7 @@ import {
   defineComponent,
   getCurrentInstance,
   nextTick,
+  onEffectCleanup,
   reactive,
   ref,
   watch,
@@ -393,6 +394,35 @@ describe('api: watch', () => {
     expect(cleanup).toHaveBeenCalledTimes(2)
   })
 
+  it('onEffectCleanup', async () => {
+    const count = ref(0)
+    const cleanupEffect = vi.fn()
+    const cleanupWatch = vi.fn()
+
+    const stopEffect = watchEffect(() => {
+      onEffectCleanup(cleanupEffect)
+      count.value
+    })
+    const stopWatch = watch(count, () => {
+      onEffectCleanup(cleanupWatch)
+    })
+
+    count.value++
+    await nextTick()
+    expect(cleanupEffect).toHaveBeenCalledTimes(1)
+    expect(cleanupWatch).toHaveBeenCalledTimes(0)
+
+    count.value++
+    await nextTick()
+    expect(cleanupEffect).toHaveBeenCalledTimes(2)
+    expect(cleanupWatch).toHaveBeenCalledTimes(1)
+
+    stopEffect()
+    expect(cleanupEffect).toHaveBeenCalledTimes(3)
+    stopWatch()
+    expect(cleanupWatch).toHaveBeenCalledTimes(2)
+  })
+
   it('flush timing: pre (default)', async () => {
     const count = ref(0)
     const count2 = ref(0)

packages/runtime-core/src/apiWatch.ts

@@ -1,27 +1,22 @@
 import {
+  type BaseWatchErrorCodes,
+  type BaseWatchOptions,
   type ComputedRef,
   type DebuggerOptions,
-  type EffectScheduler,
-  ReactiveEffect,
-  ReactiveFlags,
   type Ref,
+  baseWatch,
   getCurrentScope,
-  isReactive,
-  isRef,
-  isShallow,
 } from '@vue/reactivity'
-import { type SchedulerJob, queueJob } from './scheduler'
+import {
+  type SchedulerFactory,
+  createPreScheduler,
+  createSyncScheduler,
+} from './scheduler'
 import {
   EMPTY_OBJ,
   NOOP,
   extend,
-  hasChanged,
-  isArray,
   isFunction,
-  isMap,
-  isObject,
-  isPlainObject,
-  isSet,
   isString,
   remove,
 } from '@vue/shared'
@@ -32,15 +27,9 @@ import {
   setCurrentInstance,
   unsetCurrentInstance,
 } from './component'
-import {
-  ErrorCodes,
-  callWithAsyncErrorHandling,
-  callWithErrorHandling,
-} from './errorHandling'
-import { queuePostRenderEffect } from './renderer'
+import { handleError as handleErrorWithInstance } from './errorHandling'
+import { createPostRenderScheduler } from './renderer'
 import { warn } from './warning'
-import { DeprecationTypes } from './compat/compatConfig'
-import { checkCompatEnabled, isCompatEnabled } from './compat/compatConfig'
 import type { ObjectWatchOptionItem } from './componentOptions'
 import { useSSRContext } from './helpers/useSsrContext'
 
@@ -110,9 +99,6 @@ export function watchSyncEffect(
   )
 }
 
-// initial value for watchers to trigger on undefined initial values
-const INITIAL_WATCHER_VALUE = {}
-
 type MultiWatchSources = (WatchSource<unknown> | object)[]
 
 // overload: single source + cb
@@ -170,25 +156,23 @@ export function watch<T = any, Immediate extends Readonly<boolean> = false>(
   return doWatch(source as any, cb, options)
 }
 
+function getScheduler(flush: WatchOptionsBase['flush']): SchedulerFactory {
+  if (flush === 'post') {
+    return createPostRenderScheduler
+  }
+  if (flush === 'sync') {
+    return createSyncScheduler
+  }
+  // default: 'pre'
+  return createPreScheduler
+}
+
 function doWatch(
   source: WatchSource | WatchSource[] | WatchEffect | object,
   cb: WatchCallback | null,
-  {
-    immediate,
-    deep,
-    flush,
-    once,
-    onTrack,
-    onTrigger,
-  }: WatchOptions = EMPTY_OBJ,
+  options: WatchOptions = EMPTY_OBJ,
 ): WatchStopHandle {
-  if (cb && once) {
-    const _cb = cb
-    cb = (...args) => {
-      _cb(...args)
-      unwatch()
-    }
-  }
+  const { immediate, deep, flush, once } = options
 
   // TODO remove in 3.5
   if (__DEV__ && deep !== void 0 && typeof deep === 'number') {
@@ -219,210 +203,40 @@ function doWatch(
     }
   }
 
-  const warnInvalidSource = (s: unknown) => {
-    warn(
-      `Invalid watch source: `,
-      s,
-      `A watch source can only be a getter/effect function, a ref, ` +
-        `a reactive object, or an array of these types.`,
-    )
-  }
+  const extendOptions: BaseWatchOptions = {}
 
-  const instance = currentInstance
-  const reactiveGetter = (source: object) =>
-    deep === true
-      ? source // traverse will happen in wrapped getter below
-      : // for deep: false, only traverse root-level properties
-        traverse(source, deep === false ? 1 : undefined)
+  if (__DEV__) extendOptions.onWarn = warn
 
-  let getter: () => any
-  let forceTrigger = false
-  let isMultiSource = false
-
-  if (isRef(source)) {
-    getter = () => source.value
-    forceTrigger = isShallow(source)
-  } else if (isReactive(source)) {
-    getter = () => reactiveGetter(source)
-    forceTrigger = true
-  } else if (isArray(source)) {
-    isMultiSource = true
-    forceTrigger = source.some(s => isReactive(s) || isShallow(s))
-    getter = () =>
-      source.map(s => {
-        if (isRef(s)) {
-          return s.value
-        } else if (isReactive(s)) {
-          return reactiveGetter(s)
-        } else if (isFunction(s)) {
-          return callWithErrorHandling(s, instance, ErrorCodes.WATCH_GETTER)
-        } else {
-          __DEV__ && warnInvalidSource(s)
-        }
-      })
-  } else if (isFunction(source)) {
-    if (cb) {
-      // getter with cb
-      getter = () =>
-        callWithErrorHandling(source, instance, ErrorCodes.WATCH_GETTER)
-    } else {
-      // no cb -> simple effect
-      getter = () => {
-        if (cleanup) {
-          cleanup()
-        }
-        return callWithAsyncErrorHandling(
-          source,
-          instance,
-          ErrorCodes.WATCH_CALLBACK,
-          [onCleanup],
-        )
-      }
-    }
-  } else {
-    getter = NOOP
-    __DEV__ && warnInvalidSource(source)
-  }
-
-  // 2.x array mutation watch compat
-  if (__COMPAT__ && cb && !deep) {
-    const baseGetter = getter
-    getter = () => {
-      const val = baseGetter()
-      if (
-        isArray(val) &&
-        checkCompatEnabled(DeprecationTypes.WATCH_ARRAY, instance)
-      ) {
-        traverse(val)
-      }
-      return val
-    }
-  }
-
-  if (cb && deep) {
-    const baseGetter = getter
-    getter = () => traverse(baseGetter())
-  }
-
-  let cleanup: (() => void) | undefined
-  let onCleanup: OnCleanup = (fn: () => void) => {
-    cleanup = effect.onStop = () => {
-      callWithErrorHandling(fn, instance, ErrorCodes.WATCH_CLEANUP)
-      cleanup = effect.onStop = undefined
-    }
-  }
-
-  // in SSR there is no need to setup an actual effect, and it should be noop
-  // unless it's eager or sync flush
   let ssrCleanup: (() => void)[] | undefined
   if (__SSR__ && isInSSRComponentSetup) {
-    // we will also not call the invalidate callback (+ runner is not set up)
-    onCleanup = NOOP
-    if (!cb) {
-      getter()
-    } else if (immediate) {
-      callWithAsyncErrorHandling(cb, instance, ErrorCodes.WATCH_CALLBACK, [
-        getter(),
-        isMultiSource ? [] : undefined,
-        onCleanup,
-      ])
-    }
     if (flush === 'sync') {
       const ctx = useSSRContext()!
       ssrCleanup = ctx.__watcherHandles || (ctx.__watcherHandles = [])
+    } else if (!cb || immediate) {
+      // immediately watch or watchEffect
+      extendOptions.once = true
     } else {
+      // watch(source, cb)
       return NOOP
     }
   }
 
-  let oldValue: any = isMultiSource
-    ? new Array((source as []).length).fill(INITIAL_WATCHER_VALUE)
-    : INITIAL_WATCHER_VALUE
-  const job: SchedulerJob = () => {
-    if (!effect.active || !effect.dirty) {
-      return
-    }
-    if (cb) {
-      // watch(source, cb)
-      const newValue = effect.run()
-      if (
-        deep ||
-        forceTrigger ||
-        (isMultiSource
-          ? (newValue as any[]).some((v, i) => hasChanged(v, oldValue[i]))
-          : hasChanged(newValue, oldValue)) ||
-        (__COMPAT__ &&
-          isArray(newValue) &&
-          isCompatEnabled(DeprecationTypes.WATCH_ARRAY, instance))
-      ) {
-        // cleanup before running cb again
-        if (cleanup) {
-          cleanup()
-        }
-        callWithAsyncErrorHandling(cb, instance, ErrorCodes.WATCH_CALLBACK, [
-          newValue,
-          // pass undefined as the old value when it's changed for the first time
-          oldValue === INITIAL_WATCHER_VALUE
-            ? undefined
-            : isMultiSource && oldValue[0] === INITIAL_WATCHER_VALUE
-              ? []
-              : oldValue,
-          onCleanup,
-        ])
-        oldValue = newValue
-      }
-    } else {
-      // watchEffect
-      effect.run()
-    }
-  }
+  const instance = currentInstance
+  extendOptions.onError = (err: unknown, type: BaseWatchErrorCodes) =>
+    handleErrorWithInstance(err, instance, type)
+  extendOptions.scheduler = getScheduler(flush)(instance)
 
-  // important: mark the job as a watcher callback so that scheduler knows
-  // it is allowed to self-trigger (#1727)
-  job.allowRecurse = !!cb
-
-  let scheduler: EffectScheduler
-  if (flush === 'sync') {
-    scheduler = job as any // the scheduler function gets called directly
-  } else if (flush === 'post') {
-    scheduler = () => queuePostRenderEffect(job, instance && instance.suspense)
-  } else {
-    // default: 'pre'
-    job.pre = true
-    if (instance) job.id = instance.uid
-    scheduler = () => queueJob(job)
-  }
-
-  const effect = new ReactiveEffect(getter, NOOP, scheduler)
+  let effect = baseWatch(source, cb, extend({}, options, extendOptions))
 
   const scope = getCurrentScope()
-  const unwatch = () => {
-    effect.stop()
-    if (scope) {
-      remove(scope.effects, effect)
-    }
-  }
-
-  if (__DEV__) {
-    effect.onTrack = onTrack
-    effect.onTrigger = onTrigger
-  }
-
-  // initial run
-  if (cb) {
-    if (immediate) {
-      job()
-    } else {
-      oldValue = effect.run()
-    }
-  } else if (flush === 'post') {
-    queuePostRenderEffect(
-      effect.run.bind(effect),
-      instance && instance.suspense,
-    )
-  } else {
-    effect.run()
-  }
+  const unwatch = !effect
+    ? NOOP
+    : () => {
+        effect!.stop()
+        if (scope) {
+          remove(scope.effects, effect)
+        }
+      }
 
   if (__SSR__ && ssrCleanup) ssrCleanup.push(unwatch)
   return unwatch
@@ -469,43 +283,3 @@ export function createPathGetter(ctx: any, path: string) {
     return cur
   }
 }
-
-export function traverse(
-  value: unknown,
-  depth?: number,
-  currentDepth = 0,
-  seen?: Set<unknown>,
-) {
-  if (!isObject(value) || (value as any)[ReactiveFlags.SKIP]) {
-    return value
-  }
-
-  if (depth && depth > 0) {
-    if (currentDepth >= depth) {
-      return value
-    }
-    currentDepth++
-  }
-
-  seen = seen || new Set()
-  if (seen.has(value)) {
-    return value
-  }
-  seen.add(value)
-  if (isRef(value)) {
-    traverse(value.value, depth, currentDepth, seen)
-  } else if (isArray(value)) {
-    for (let i = 0; i < value.length; i++) {
-      traverse(value[i], depth, currentDepth, seen)
-    }
-  } else if (isSet(value) || isMap(value)) {
-    value.forEach((v: any) => {
-      traverse(v, depth, currentDepth, seen)
-    })
-  } else if (isPlainObject(value)) {
-    for (const key in value) {
-      traverse(value[key], depth, currentDepth, seen)
-    }
-  }
-  return value
-}

packages/runtime-core/src/componentOptions.ts

@@ -1,10 +1,11 @@
-import type {
-  Component,
-  ComponentInternalInstance,
-  ComponentInternalOptions,
-  ConcreteComponent,
-  InternalRenderFunction,
-  SetupContext,
+import {
+  type Component,
+  type ComponentInternalInstance,
+  type ComponentInternalOptions,
+  type ConcreteComponent,
+  type InternalRenderFunction,
+  type SetupContext,
+  currentInstance,
 } from './component'
 import {
   type Data,
@@ -18,7 +19,7 @@ import {
   isPromise,
   isString,
 } from '@vue/shared'
-import { type Ref, isRef } from '@vue/reactivity'
+import { type Ref, getCurrentScope, isRef, traverse } from '@vue/reactivity'
 import { computed } from './apiComputed'
 import {
   type WatchCallback,
@@ -67,7 +68,7 @@ import { warn } from './warning'
 import type { VNodeChild } from './vnode'
 import { callWithAsyncErrorHandling } from './errorHandling'
 import { deepMergeData } from './compat/data'
-import { DeprecationTypes } from './compat/compatConfig'
+import { DeprecationTypes, checkCompatEnabled } from './compat/compatConfig'
 import {
   type CompatConfig,
   isCompatEnabled,
@@ -937,18 +938,45 @@ export function createWatcher(
   publicThis: ComponentPublicInstance,
   key: string,
 ) {
-  const getter = key.includes('.')
+  let getter = key.includes('.')
     ? createPathGetter(publicThis, key)
     : () => (publicThis as any)[key]
+
+  const options: WatchOptions = {}
+  if (__COMPAT__) {
+    const instance =
+      getCurrentScope() === currentInstance?.scope ? currentInstance : null
+
+    const newValue = getter()
+    if (
+      isArray(newValue) &&
+      isCompatEnabled(DeprecationTypes.WATCH_ARRAY, instance)
+    ) {
+      options.deep = true
+    }
+
+    const baseGetter = getter
+    getter = () => {
+      const val = baseGetter()
+      if (
+        isArray(val) &&
+        checkCompatEnabled(DeprecationTypes.WATCH_ARRAY, instance)
+      ) {
+        traverse(val)
+      }
+      return val
+    }
+  }
+
   if (isString(raw)) {
     const handler = ctx[raw]
     if (isFunction(handler)) {
-      watch(getter, handler as WatchCallback)
+      watch(getter, handler as WatchCallback, options)
     } else if (__DEV__) {
       warn(`Invalid watch handler specified by key "${raw}"`, handler)
     }
   } else if (isFunction(raw)) {
-    watch(getter, raw.bind(publicThis))
+    watch(getter, raw.bind(publicThis), options)
   } else if (isObject(raw)) {
     if (isArray(raw)) {
       raw.forEach(r => createWatcher(r, ctx, publicThis, key))
@@ -957,7 +985,7 @@ export function createWatcher(
         ? raw.handler.bind(publicThis)
         : (ctx[raw.handler] as WatchCallback)
       if (isFunction(handler)) {
-        watch(getter, handler, raw)
+        watch(getter, handler, extend(raw, options))
       } else if (__DEV__) {
         warn(`Invalid watch handler specified by key "${raw.handler}"`, handler)
       }

packages/runtime-core/src/directives.ts

@@ -24,8 +24,7 @@ import { currentRenderingInstance } from './componentRenderContext'
 import { ErrorCodes, callWithAsyncErrorHandling } from './errorHandling'
 import type { ComponentPublicInstance } from './componentPublicInstance'
 import { mapCompatDirectiveHook } from './compat/customDirective'
-import { pauseTracking, resetTracking } from '@vue/reactivity'
-import { traverse } from './apiWatch'
+import { pauseTracking, resetTracking, traverse } from '@vue/reactivity'
 
 export interface DirectiveBinding<V = any> {
   instance: ComponentPublicInstance | null

packages/runtime-core/src/errorHandling.ts

@@ -3,16 +3,20 @@ import type { ComponentInternalInstance } from './component'
 import { popWarningContext, pushWarningContext, warn } from './warning'
 import { isFunction, isPromise } from '@vue/shared'
 import { LifecycleHooks } from './enums'
+import { BaseWatchErrorCodes } from '@vue/reactivity'
 
 // contexts where user provided function may be executed, in addition to
 // lifecycle hooks.
 export enum ErrorCodes {
   SETUP_FUNCTION,
   RENDER_FUNCTION,
-  WATCH_GETTER,
-  WATCH_CALLBACK,
-  WATCH_CLEANUP,
-  NATIVE_EVENT_HANDLER,
+  // The error codes for the watch have been transferred to the reactivity
+  // package along with baseWatch to maintain code compatibility. Hence,
+  // it is essential to keep these values unchanged.
+  // WATCH_GETTER,
+  // WATCH_CALLBACK,
+  // WATCH_CLEANUP,
+  NATIVE_EVENT_HANDLER = 5,
   COMPONENT_EVENT_HANDLER,
   VNODE_HOOK,
   DIRECTIVE_HOOK,
@@ -24,7 +28,9 @@ export enum ErrorCodes {
   SCHEDULER,
 }
 
-export const ErrorTypeStrings: Record<LifecycleHooks | ErrorCodes, string> = {
+export type ErrorTypes = LifecycleHooks | ErrorCodes | BaseWatchErrorCodes
+
+export const ErrorTypeStrings: Record<ErrorTypes, string> = {
   [LifecycleHooks.SERVER_PREFETCH]: 'serverPrefetch hook',
   [LifecycleHooks.BEFORE_CREATE]: 'beforeCreate hook',
   [LifecycleHooks.CREATED]: 'created hook',
@@ -41,9 +47,9 @@ export const ErrorTypeStrings: Record<LifecycleHooks | ErrorCodes, string> = {
   [LifecycleHooks.RENDER_TRIGGERED]: 'renderTriggered hook',
   [ErrorCodes.SETUP_FUNCTION]: 'setup function',
   [ErrorCodes.RENDER_FUNCTION]: 'render function',
-  [ErrorCodes.WATCH_GETTER]: 'watcher getter',
-  [ErrorCodes.WATCH_CALLBACK]: 'watcher callback',
-  [ErrorCodes.WATCH_CLEANUP]: 'watcher cleanup function',
+  [BaseWatchErrorCodes.WATCH_GETTER]: 'watcher getter',
+  [BaseWatchErrorCodes.WATCH_CALLBACK]: 'watcher callback',
+  [BaseWatchErrorCodes.WATCH_CLEANUP]: 'watcher cleanup function',
   [ErrorCodes.NATIVE_EVENT_HANDLER]: 'native event handler',
   [ErrorCodes.COMPONENT_EVENT_HANDLER]: 'component event handler',
   [ErrorCodes.VNODE_HOOK]: 'vnode hook',
@@ -58,8 +64,6 @@ export const ErrorTypeStrings: Record<LifecycleHooks | ErrorCodes, string> = {
     'Please open an issue at https://github.com/vuejs/core .',
 }
 
-export type ErrorTypes = LifecycleHooks | ErrorCodes
-
 export function callWithErrorHandling(
   fn: Function,
   instance: ComponentInternalInstance | null,

packages/runtime-core/src/index.ts

@@ -28,6 +28,7 @@ export {
   // effect
   effect,
   stop,
+  onEffectCleanup,
   ReactiveEffect,
   // effect scope
   effectScope,

packages/runtime-core/src/renderer.ts

@@ -38,6 +38,7 @@ import {
   isReservedProp,
 } from '@vue/shared'
 import {
+  type SchedulerFactory,
   type SchedulerJob,
   flushPostFlushCbs,
   flushPreFlushCbs,
@@ -281,6 +282,18 @@ export const queuePostRenderEffect = __FEATURE_SUSPENSE__
     : queueEffectWithSuspense
   : queuePostFlushCb
 
+export const createPostRenderScheduler: SchedulerFactory =
+  instance => (job, effect, isInit) => {
+    if (isInit) {
+      queuePostRenderEffect(
+        effect.run.bind(effect),
+        instance && instance.suspense,
+      )
+    } else {
+      queuePostRenderEffect(job, instance && instance.suspense)
+    }
+  }
+
 /**
  * The createRenderer function accepts two generic arguments:
  * HostNode and HostElement, corresponding to Node and Element types in the

packages/runtime-core/src/scheduler.ts

@@ -1,6 +1,7 @@
 import { ErrorCodes, callWithErrorHandling, handleError } from './errorHandling'
 import { type Awaited, NOOP, isArray } from '@vue/shared'
 import { type ComponentInternalInstance, getComponentName } from './component'
+import type { Scheduler } from '@vue/reactivity'
 
 export interface SchedulerJob extends Function {
   id?: number
@@ -287,3 +288,27 @@ function checkRecursiveUpdates(seen: CountMap, fn: SchedulerJob) {
     }
   }
 }
+
+export type SchedulerFactory = (
+  instance: ComponentInternalInstance | null,
+) => Scheduler
+
+export const createSyncScheduler: SchedulerFactory =
+  instance => (job, effect, isInit) => {
+    if (isInit) {
+      effect.run()
+    } else {
+      job()
+    }
+  }
+
+export const createPreScheduler: SchedulerFactory =
+  instance => (job, effect, isInit) => {
+    if (isInit) {
+      effect.run()
+    } else {
+      job.pre = true
+      if (instance) job.id = instance.uid
+      queueJob(job)
+    }
+  }