diff --git a/.changeset/atom-registry-stale-dependencies.md b/.changeset/atom-registry-stale-dependencies.md new file mode 100644 index 00000000000..ab14a026ebe --- /dev/null +++ b/.changeset/atom-registry-stale-dependencies.md @@ -0,0 +1,5 @@ +--- +"effect": patch +--- + +Fix missed updates through stale `AtomRegistry` nodes, premature dependency release, leaked superseded builds, and failed observed builds that never recover. diff --git a/packages/effect/src/reactivity/AtomRegistry.ts b/packages/effect/src/reactivity/AtomRegistry.ts index dcda89f6dcb..91b1d00025e 100644 --- a/packages/effect/src/reactivity/AtomRegistry.ts +++ b/packages/effect/src/reactivity/AtomRegistry.ts @@ -323,8 +323,24 @@ export const mount: { const constImmediate = { immediate: true } -const notifyListener = (listener: () => void): void => { - listener() +const invoke = (f: () => void): void => { + f() +} + +const noFailure = {} + +// Runs `f` for every item, then rethrows the first failure. One failing step +// must not skip the others: those would leak a lifetime or miss an update. +function runAll(items: Iterable, f: (item: T) => void): void { + let failure: unknown = noFailure + for (const item of items) { + try { + f(item) + } catch (error) { + if (failure === noFailure) failure = error + } + } + if (failure !== noFailure) throw failure } const SerializableTypeId: Atom.SerializableTypeId = "~effect-atom/atom/Atom/Serializable" @@ -415,6 +431,9 @@ class RegistryImpl implements AtomRegistry { const node = this.ensureNode(atom) if (options?.immediate) { f(node.value()) + } else if (node.state === NodeState.stale) { + // a listener hears the next change, which only reaches an up-to-date node + node.value() } const remove = node.subscribe(function() { f(node._value) @@ -553,16 +572,20 @@ class RegistryImpl implements AtomRegistry { sweepBucket(bucket: number): void { const nodes = this.timeoutBuckets.get(bucket)![0] this.timeoutBuckets.delete(bucket) + runAll(nodes, this.removeIdleNode) + } - nodes.forEach((node) => { - this.nodeTimeoutBucket.delete(node) - if (!node.canBeRemoved) return - this.nodes.delete(atomKey(node.atom)) - this.onNodeRemoved?.(node) - this.#currentSweepTTL = node.atom.idleTTL ?? this.defaultIdleTTL! + removeIdleNode = (node: NodeImpl): void => { + this.nodeTimeoutBucket.delete(node) + if (!node.canBeRemoved) return + this.nodes.delete(atomKey(node.atom)) + this.onNodeRemoved?.(node) + this.#currentSweepTTL = node.atom.idleTTL ?? this.defaultIdleTTL! + try { node.remove() + } finally { this.#currentSweepTTL = null - }) + } } reset(): void { @@ -570,11 +593,18 @@ class RegistryImpl implements AtomRegistry { this.timeoutBuckets.clear() this.nodeTimeoutBucket.clear() - this.nodes.forEach((node) => { - node.remove() - this.onNodeRemoved?.(node) - }) + const nodes = Array.from(this.nodes.values()) this.nodes.clear() + // Mark all nodes before removing any, so a parent cannot be rescheduled for removal. + for (const node of nodes) { + node.state = NodeState.removed + } + runAll(nodes, this.destroyNode) + } + + destroyNode = (node: NodeImpl): void => { + node.remove() + this.onNodeRemoved?.(node) } dispose(): void { @@ -614,14 +644,16 @@ class NodeImpl { lifetime: Lifetime | undefined writeContext: WriteContextImpl preserveInitialValueOnBuild = false + invalidatedDuringBuild = false + failed = false + // Every parent that holds this node as a child. A stale node keeps the + // parents of its last build, so they are not released and their next change + // still reaches it. A build drops the parents it did not read. parents = new Set>() - previousParents: Set> | undefined children = new Set>() listeners = new Set<() => void>() - skipInvalidation = false - building = false - invalidatedDuringBuild = false + observedChildren = 0 currentState() { switch (this.state) { @@ -640,35 +672,128 @@ class NodeImpl { return !this.atom.keepAlive && this.listeners.size === 0 && this.children.size === 0 && this.state !== 0 } + get building(): boolean { + return this.lifetime !== undefined && this.lifetime.reads !== undefined + } + + // An observed node is rebuilt as soon as it goes stale, because something + // waits for its value without reading it. The count spares walking the + // children on every invalidation. + get isObserved(): boolean { + return !this.atom.lazy || this.listeners.size > 0 || this.observedChildren > 0 + } + + childObserved(delta: 1 | -1): void { + const observed = this.isObserved + this.observedChildren += delta + if (observed !== this.isObserved) { + this.observedChanged(delta) + } + } + + observedChanged(delta: 1 | -1): void { + for (const parent of this.parents) { + parent.childObserved(delta) + } + } + _value: A = undefined as any value(): A { - if ((this.state & NodeFlags.waitingForValue) !== 0) { - this.lifetime = makeLifetime(this) - this.building = true - const value = this.atom.read(this.lifetime) - this.building = false - if ((this.state & NodeFlags.waitingForValue) !== 0) { - if (this.preserveInitialValueOnBuild) { - this.preserveInitialValueOnBuild = false - this.state = NodeState.valid - } else { - this.setValue(value) - } + // a build can be superseded by a nested one, and that one staled again + // before the outer read returns + while ((this.state & NodeFlags.waitingForValue) !== 0) { + this.build() + } + return this._value + } + + build(): void { + // a lifetime still attached to a node that needs a value belongs to a + // build that failed, or to one that is running and is now superseded + this.disposeLifetime() + this.invalidatedDuringBuild = false + const lifetime = makeLifetime(this) + this.lifetime = lifetime + let value: A + let read = false + try { + value = this.atom.read(lifetime) + read = true + } finally { + const reads = lifetime.reads! + lifetime.reads = undefined + if (this.lifetime === lifetime) { + this.dropUnread(reads) + if (!read) this.fail() + } + } + if (this.lifetime === lifetime && (this.state & NodeFlags.waitingForValue) !== 0) { + if (this.preserveInitialValueOnBuild) { + this.preserveInitialValueOnBuild = false + this.failed = false + this.state = NodeState.valid + } else { + this.setValue(value) } + } + if (lifetime.failure !== noFailure) throw lifetime.failure + } - if (this.previousParents) { - const parents = this.previousParents - this.previousParents = undefined - for (const parent of parents) { - parent.removeChild(this) - if (parent.canBeRemoved) { - this.registry.scheduleNodeRemoval(parent) - } - } + // A build that threw leaves its dependents holding, or reading, a value + // this node no longer stands behind, so they go stale and rebuild when + // pulled or observed. The read's error is the one reported. + fail(): void { + this.failed = true + try { + this.invalidateChildren() + } catch { + // an observed dependent rebuilt and met the same failure + } + } + + readTracked(atom: Atom.Atom, lifetime: Lifetime): B { + const parent = this.registry.ensureNode(atom) + // the dependency holds even when the parent cannot produce a value, so + // that its recovery reaches this node; a build superseded while the + // parent builds has the edge dropped by its replacement + if (!this.parents.has(parent)) { + this.parents.add(parent) + parent.children.add(this) + if (this.isObserved) { + parent.childObserved(1) + } + } + try { + return parent.value() + } catch (error) { + // the parent may have got its value and failed only to update another + // dependent: this build still takes the value, and rethrows once it is done + if (lifetime.reads === undefined || parent.state !== NodeState.valid) throw error + if (lifetime.failure === noFailure) lifetime.failure = error + return parent._value + } finally { + // read once the parent has answered: its commit while it is being + // pulled is the value this build is about to receive + if (lifetime.reads !== undefined && !lifetime.disposed) { + lifetime.reads.add(parent) } } + } - return this._value + dropUnread(reads: Set>): void { + if (reads.size === this.parents.size) return + const observed = this.isObserved + for (const parent of this.parents) { + if (reads.has(parent)) continue + this.parents.delete(parent) + parent.children.delete(this) + if (observed) { + parent.childObserved(-1) + } + if (parent.canBeRemoved) { + this.registry.scheduleNodeRemoval(parent) + } + } } valueOption(): Option.Option { @@ -683,13 +808,9 @@ class NodeImpl { this.preserveInitialValueOnBuild = true this.state = NodeState.stale this._value = value - - if (batchState.phase === BatchPhase.collect) { - batchState.notify.add(this) - } else { - this.notify() + if (this.listeners.size > 0) { + this.announce() } - return } @@ -697,77 +818,68 @@ class NodeImpl { } setValue(value: A): void { - if ((this.state & NodeFlags.initialized) === 0) { - this.state = NodeState.valid - this._value = value - - if (batchState.phase === BatchPhase.collect) { - batchState.notify.add(this) - } else { - this.notify() - } - - return - } - + // dependents that failed along with this node get another try + const recovered = this.failed + this.failed = false + const initialized = (this.state & NodeFlags.initialized) !== 0 this.state = NodeState.valid - if (this.atom.equals(this._value, value)) { + if (initialized && this.atom.equals(this._value, value)) { + if (recovered) this.invalidateChildren() return } this._value = value - if (this.skipInvalidation) { - this.skipInvalidation = false - } else { + let failure: unknown = noFailure + try { this.invalidateChildren() + } catch (error) { + failure = error } - + // the value is committed, so its listeners hear of it even when a dependent failed if (this.listeners.size > 0) { - if (batchState.phase === BatchPhase.collect) { - batchState.notify.add(this) - } else { - this.notify() + try { + this.announce() + } catch (error) { + if (failure === noFailure) failure = error } } + if (failure !== noFailure) throw failure } - addParent(parent: NodeImpl): void { - this.parents.add(parent) - if (this.previousParents !== undefined) { - this.previousParents.delete(parent) - if (this.previousParents.size === 0) { - this.previousParents = undefined - } - } - - if (!parent.children.has(this)) { - parent.children.add(this) - if (parent.skipInvalidation) { - parent.skipInvalidation = false - } + announce(): void { + if (batchState.phase === BatchPhase.collect) { + batchState.notify.add(this) + } else { + this.notify() } } - removeChild(child: NodeImpl): void { - this.children.delete(child) - } - invalidate(): void { - if (this.building && batchState.phase === BatchPhase.collect) { - this.invalidatedDuringBuild = true - } if (this.state === NodeState.valid) { this.state = NodeState.stale - this.disposeLifetime() + try { + this.disposeLifetime() + } finally { + this.propagate() + } + } else if (this.building && batchState.phase !== BatchPhase.collect) { + // the running build read a parent that has changed since + this.build() + } else { + this.propagate() } + } + propagate(): void { if (batchState.phase === BatchPhase.collect) { + if (this.building) { + this.invalidatedDuringBuild = true + } batchState.stale.push(this) - } else if (this.atom.lazy && this.listeners.size === 0 && !childrenAreActive(this.children)) { - this.invalidateChildren() - this.skipInvalidation = true - } else { + } else if (this.isObserved) { this.value() + } else { + this.invalidateChildren() } } @@ -775,95 +887,94 @@ class NodeImpl { if (this.children.size === 0) { return } - - const children = this.children - this.children = new Set() - for (const child of children) { - child.invalidate() + let failure: unknown = noFailure + // a snapshot: a rebuild adds and drops children + for (const child of Array.from(this.children)) { + if (!child.affectedBy(this)) continue + try { + child.invalidate() + } catch (error) { + if (failure === noFailure) failure = error + } } + if (failure !== noFailure) throw failure + } + + // Whether a change of the parent reaches this node: it holds the parent's + // value, or a running build has already read it. A stale node is stale + // already, and a build that has not read the parent yet will read the new + // value. The exception is an observed node that is still stale: its rebuild + // failed, and a parent that has a value again is its chance to recover. + affectedBy(parent: NodeImpl): boolean { + if (this.state === NodeState.valid) return true + const reads = this.lifetime?.reads + if (reads !== undefined) return reads.has(parent) + return this.isObserved && parent.state === NodeState.valid } notify(): void { if (batchState.phase === BatchPhase.commit) { batchState.notify.delete(this) } - - this.listeners.forEach(notifyListener) + runAll(this.listeners, invoke) } disposeLifetime(): void { - if (this.lifetime !== undefined) { - this.lifetime.dispose() + const lifetime = this.lifetime + if (lifetime !== undefined) { this.lifetime = undefined - } - - if (this.parents.size !== 0) { - this.previousParents = this.parents - this.parents = new Set() + lifetime.dispose() } } remove() { + const observed = this.isObserved this.state = NodeState.removed this.listeners.clear() - - if (this.lifetime !== undefined) { + try { this.disposeLifetime() - } - - if (this.previousParents === undefined) { - return - } - - const parents = this.previousParents - this.previousParents = undefined - for (const parent of parents) { - parent.removeChild(this) - if (parent.canBeRemoved) { - this.registry.removeNode(parent) - } + } finally { + const parents = Array.from(this.parents) + this.parents.clear() + runAll(parents, (parent) => { + parent.children.delete(this) + if (observed) { + parent.childObserved(-1) + } + if (parent.canBeRemoved) { + this.registry.removeNode(parent) + } + }) } } subscribe(listener: () => void): () => void { + const observed = this.isObserved this.listeners.add(listener) - return () => this.listeners.delete(listener) - } -} - -function childrenAreActive(children: Set>): boolean { - if (children.size === 0) { - return false - } - let current: Set> | undefined = children - let stack: Array>> | undefined - let stackIndex = 0 - while (current !== undefined) { - for (const child of current) { - if (!child.atom.lazy || child.listeners.size > 0) { - return true - } else if (child.children.size > 0) { - if (stack === undefined) { - stack = [child.children] - } else { - stack.push(child.children) - } + if (!observed) { + this.observedChanged(1) + } + return () => { + if (this.listeners.delete(listener) && !this.isObserved) { + this.observedChanged(-1) } } - current = stack?.[stackIndex++] } - return false } interface Lifetime extends Atom.AtomContext { isFn: boolean readonly node: NodeImpl + // the parents read so far by the build; `undefined` once the build has returned + reads: Set> | undefined + // a failure met while reading a parent that still produced its value + failure: unknown finalizers: Array<() => void> | undefined disposed: boolean readonly dispose: () => void } -const LifetimeProto: Omit, "node" | "finalizers" | "disposed" | "isFn"> = { +const LifetimeProto: Omit, "node" | "finalizers" | "disposed" | "isFn" | "reads" | "failure"> = { get registry(): RegistryImpl { return (this as Lifetime).node.registry }, @@ -878,10 +989,7 @@ const LifetimeProto: Omit, "node" | "finalizers" | "disposed" | "i if (this.disposed) { return this.node.registry.get(atom) } - const parent = this.node.registry.ensureNode(atom) - const value = parent.value() - this.node.addParent(parent) - return value + return this.node.readTracked(atom, this) }, result(this: Lifetime, atom: Atom.Atom>, options?: { @@ -1031,29 +1139,24 @@ const LifetimeProto: Omit, "node" | "finalizers" | "disposed" | "i const finalizers = this.finalizers this.finalizers = undefined - for (let i = finalizers.length - 1; i >= 0; i--) { - finalizers[i]() - } + runAll(finalizers.reverse(), invoke) } } const makeLifetime = (node: NodeImpl): Lifetime => { function get(atom: Atom.Atom): A { - if (get.disposed) { - return node.registry.get(atom) - } else if (get.isFn) { + if (get.disposed || get.isFn) { return node.registry.get(atom) } - const parent = node.registry.ensureNode(atom) - const value = parent.value() - node.addParent(parent) - return value + return node.readTracked(atom, get as any) } Object.setPrototypeOf(get, LifetimeProto) get.isFn = false get.disposed = false get.finalizers = undefined get.node = node + get.reads = new Set>() + get.failure = noFailure return get as any } @@ -1116,29 +1219,40 @@ export function batch(f: () => void): void { const previousPhase = batchState.phase batchState.phase = BatchPhase.collect batchState.depth++ + let failure: unknown = noFailure try { f() - if (batchState.depth === 1) { - let i = 0 - do { - batchState.phase = BatchPhase.collect - for (; i < batchState.stale.length; i++) { - batchRebuildNode(batchState.stale[i]) + } catch (error) { + failure = error + } + if (batchState.depth === 1) { + // what `f` changed before it failed is committed, so it is still rebuilt and heard + const { notify, stale } = batchState + let i = 0 + do { + batchState.phase = BatchPhase.collect + for (; i < stale.length; i++) { + try { + batchRebuildNode(stale[i]) + } catch (error) { + if (failure === noFailure) failure = error } - batchState.phase = BatchPhase.commit - for (const node of batchState.notify) { + } + batchState.phase = BatchPhase.commit + for (const node of notify) { + try { node.notify() + } catch (error) { + if (failure === noFailure) failure = error } - } while (i < batchState.stale.length) - } - } finally { - batchState.depth-- - batchState.phase = previousPhase - if (batchState.depth === 0) { - batchState.stale = [] - batchState.notify.clear() - } + } + } while (i < stale.length) + batchState.stale = [] + notify.clear() } + batchState.depth-- + batchState.phase = previousPhase + if (failure !== noFailure) throw failure } function batchRebuildNode(node: NodeImpl) { @@ -1150,15 +1264,5 @@ function batchRebuildNode(node: NodeImpl) { node.state = NodeState.stale node.disposeLifetime() } - - for (const parent of node.parents) { - if (parent.state !== NodeState.valid) { - batchRebuildNode(parent) - } - } - - // @ts-ignore - if (node.state !== NodeState.valid) { - node.value() - } + node.value() } diff --git a/packages/effect/test/reactivity/Atom.test.ts b/packages/effect/test/reactivity/Atom.test.ts index 38dc65b8192..5c8a0c43da6 100644 --- a/packages/effect/test/reactivity/Atom.test.ts +++ b/packages/effect/test/reactivity/Atom.test.ts @@ -527,6 +527,69 @@ describe("Atom", { concurrent: false }, () => { unsubscribe() }) + it("a stale dependent keeps its dependency", async () => { + const a = Atom.make(0) + let builds = 0 + const dependency = Atom.make(() => ++builds) + const dependent = Atom.make((get) => get(a) + get(dependency)).pipe(Atom.keepAlive) + const r = AtomRegistry.make() + r.get(dependent) + r.set(a, 1) + await Effect.runPromise(Effect.yieldNow) + assert.strictEqual(r.get(dependent), 2) + }) + + it("a listener added to a stale node hears changes through a parent that rebuilds to the same value", () => { + const a = Atom.make(0) + const b = Atom.make(0) + const middle = Atom.make((get) => get(a) > 0 ? 1 : 0) + const derived = Atom.make((get) => get(middle) + get(b)) + const r = AtomRegistry.make() + r.get(derived) + r.set(b, 5) + r.set(a, 1) + const seen: Array = [] + r.subscribe(derived, (value) => seen.push(value)) + r.set(a, 0) + assert.deepStrictEqual(seen, [5]) + }) + + it("recovers an observed derived atom after its build throws", () => { + const source = Atom.make(0) + const derived = Atom.make((get) => { + const value = get(source) + if (value === 1) throw new Error("build failed") + return value + }) + const r = AtomRegistry.make() + const sourceValues: Array = [] + const derivedValues: Array = [] + + assert.strictEqual(r.get(derived), 0) + r.subscribe(source, (value) => sourceValues.push(value)) + r.subscribe(derived, (value) => derivedValues.push(value)) + + assert.throws(() => r.set(source, 1), /build failed/) + assert.deepStrictEqual(sourceValues, [1]) + + r.set(source, 2) + assert.deepStrictEqual(derivedValues, [2]) + }) + + it("a build superseded while it runs is released", () => { + const p = Atom.make(0) + let finalized = 0 + const n = Atom.make((get) => { + get.addFinalizer(() => finalized++) + if (get(p) === 0) get.set(p, 1) + return get(p) + }) + const r = AtomRegistry.make() + r.subscribe(n, () => {}) + r.get(n) + assert.strictEqual(finalized, 1) + }) + it("refresh derived before mount resolves base effect", async () => { const baseAtom = Atom.make( Effect.succeed("value").pipe(Effect.delay(100)) @@ -1083,6 +1146,43 @@ describe("Atom", { concurrent: false }, () => { expect(r.get(derived)).toEqual("2b") }) + it.each([ + { existingListener: false, expected: [0] }, + { existingListener: true, expected: [0, 0] } + ])("delivers a batched value to a late subscriber (existing listener: $existingListener)", ({ + existingListener, + expected + }) => { + const r = AtomRegistry.make() + const state = Atom.make(existingListener ? 1 : 0) + const seen: Array = [] + + Atom.batch(() => { + if (existingListener) { + r.subscribe(state, () => {}) + r.set(state, 0) + } else { + r.get(state) + } + r.subscribe(state, (value) => seen.push(value), { immediate: true }) + }) + + assert.deepStrictEqual(seen, expected) + r.dispose() + }) + + it("does not queue an initialValues notification without listeners", () => { + const state = Atom.make(0) + const seen: Array = [] + + Atom.batch(() => { + const r = AtomRegistry.make({ initialValues: [Atom.initialValue(state, 10)] }) + r.subscribe(state, (value) => seen.push(value), { immediate: true }) + }) + + assert.deepStrictEqual(seen, [10]) + }) + it("runs Atom.fn writes from batch commit listeners", () => { const registry = AtomRegistry.make() const source = Atom.make(0)