fix: three concurrency holes in the item store's command layer

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-10-13 08:02:47 +00:00 · 2026-09-15 11:30:55 +02:00
parent b33f70e77f
commit 04eed3f58a
2 changed files with 126 additions and 16 deletions
@@ -195,6 +195,12 @@ class Entry<V> {
 	stopWatch: (() => void) | undefined
 	/** Fields a `patch` has put on the deployed side ahead of the server. */
 	private patched: Record<string, unknown> = {}
+	/** For each of those fields, what the server last held, so a refused patch gives back that
+	 *  and not another patch's unaccepted value. */
+	private patchedBase: Record<string, unknown> = {}
+	/** Counts rows actually handed to the syncer, so a read can tell whether one went out while
+	 *  it was in flight. */
+	private rowWrites = 0
 	/** The value as last observed, serialized: `touch` acts only on a real change. */
 	private seen: string | undefined
 	/** The row last handed to the syncer (`null`: none, `undefined`: not known — the next
@@ -252,6 +258,7 @@ class Entry<V> {
 			return
 		}
 		this.row = desired
+		this.rowWrites++
 		this.ports.write(key, desired === null ? null : snapshot(this.value))
 	}

@@ -319,13 +326,14 @@ class Entry<V> {
 		const editsAtStart = editedBefore ?? this.edits
 		return this.run(async () => {
 			const key = this.key
-			// Whether anything could have written the row while this read was in flight: only an
-			// item already on screen has a row this entry is keeping.
-			const held = this.loaded
+			const rowsAtStart = this.rowWrites
 			let res: ItemLoad<V>
 			try {
 				if (this.retired) return superseded
 				if (!adapter.load) throw new Error('This item cannot be loaded')
+				// A row queued by whoever held this key before is debounced: send it, or this read
+				// answers with the deployed value and the draft reappears under a clean editor.
+				await this.ports.flush(key)
 				res = await adapter.load(key)
 			} catch (e) {
 				this.error = errorMessage(e)
@@ -347,10 +355,10 @@ class Entry<V> {
 			}
 			this.deployed = this.withPatched(snapshot(res.deployed))
 			this.origin = res.deployed !== undefined ? 'deployed' : 'draft'
-			// A row written while this read was in flight is newer than the one it read, and the
-			// syncer has already sent it: reseeding from this response would rewind `last_sync`
-			// past that, and every write after it is refused as a conflict.
-			if (!(held && keepValue)) {
+			// A row that went out while this read was in flight is newer than the one it read:
+			// reseeding from this response would rewind `last_sync` past what the syncer has
+			// already sent, and everything after that is refused as a conflict.
+			if (this.rowWrites === rowsAtStart) {
 				this.row = serialize(res.draft) ?? null
 				this.ports.seedSync(key, res.draftSavedAt)
 			}
@@ -560,12 +568,26 @@ class Entry<V> {
 		const before = snapshot(this[baseKey]) as Record<string, unknown> | undefined
 		const sent = snapshot(fields) as Record<string, unknown>
 		if (before !== undefined) this[baseKey] = { ...before, ...sent } as V
-		if (baseKey === 'deployed') Object.assign(this.patched, sent)
+		if (baseKey === 'deployed') {
+			// The field's server value is what it held before the first patch took it, not what
+			// the patch this one overtakes put there and may yet have refused.
+			for (const k of Object.keys(sent)) {
+				if (!(k in this.patched)) this.patchedBase[k] = before?.[k]
+			}
+			Object.assign(this.patched, sent)
+		}
 		if (this.value !== undefined) Object.assign(this.value as object, fields)
 		this.touch()
-		const settled = () => {
+		/** Hand each field this patch still owns back to whoever is next: nobody if it landed as
+		 *  sent, the server's value if it was refused, a later patch otherwise. */
+		const settled = (ok: boolean) => {
 			for (const [k, v] of Object.entries(sent)) {
-				if (deepEqual(this.patched[k], v)) delete this.patched[k]
+				if (deepEqual(this.patched[k], v)) {
+					delete this.patched[k]
+					delete this.patchedBase[k]
+				} else if (ok) {
+					this.patchedBase[k] = v
+				}
 			}
 		}
 		return this.run(async () => {
@@ -573,28 +595,29 @@ class Entry<V> {
 			try {
 				await write(this.key)
 			} catch (e) {
-				settled()
+				const base = { ...this.patchedBase }
+				settled(false)
 				const giveBack = (side: V | undefined): V | undefined => {
 					if (side === undefined || before === undefined) return undefined
 					const out = snapshot(side) as Record<string, unknown>
 					let changed = false
 					for (const [k, v] of Object.entries(sent)) {
 						if (deepEqual(out[k], v)) {
-							out[k] = before[k]
+							out[k] = baseKey === 'deployed' && k in base ? base[k] : before[k]
 							changed = true
 						}
 					}
 					return changed ? (out as V) : undefined
 				}
-				const base = giveBack(this[baseKey])
-				if (base !== undefined) this[baseKey] = base
+				const rolled = giveBack(this[baseKey])
+				if (rolled !== undefined) this[baseKey] = rolled
 				const value = giveBack(this.value)
 				if (value !== undefined) this.replaceValue(value)
 				this.error = errorMessage(e)
 				this.reconcile()
 				return { ok: false, error: this.error }
 			}
-			settled()
+			settled(true)
 			this.error = undefined
 			// `settled` has dropped from `patched` whatever this patch owned, so what is left is a
 			// newer patch of the same field, whose value stands over this one's.
@@ -57,7 +57,12 @@ function fakeRows() {
 			conflicts.delete(key.path)
 			writes.push({ path: key.path, value })
 		},
-		seedSync: (key, at) => void seeds.push({ path: key.path, at }),
+		seedSync: (key, at) => {
+			seeds.push({ path: key.path, at })
+			// Back in sync with the server, as `recordRemoteSync` does.
+			conflicts.delete(key.path)
+			failures.delete(key.path)
+		},
 		conflicted: (key) => conflicts.has(key.path),
 		failure: (key) => failures.get(key.path),
 		dropPending: (key) => void dropped.push(key.path),
@@ -372,6 +377,32 @@ describe('item store: commands', () => {
 		expect(rows.writes).toEqual([])
 	})

+	it('gives the server value back when two overlapping toggles are both refused', async () => {
+		type Sched = { path: string; enabled: boolean }
+		const rows = fakeRows()
+		const store = createItemStore(rows.port)
+		const { handle: item } = store.acquire(
+			{ workspace: 'w', kind: 'trigger_schedule', path: 's' },
+			{ workspace: 'w', path: 's' },
+			{ load: async () => ({ deployed: { path: 's', enabled: false } }) } as ItemAdapter<Sched>
+		)
+		await settle()
+		const first = deferred()
+		const second = deferred()
+		const enabling = item.patch({ enabled: true }, () => first.promise)
+		const disabling = item.patch({ enabled: false }, () => second.promise)
+		first.reject(new Error('refused'))
+		second.reject(new Error('refused'))
+
+		expect(await enabling).toMatchObject({ ok: false })
+		expect(await disabling).toMatchObject({ ok: false })
+		// The second's rollback target is what the server holds, not what the first optimistically
+		// put there and never got accepted.
+		expect(item.deployed).toEqual({ path: 's', enabled: false })
+		expect(item.value).toEqual({ path: 's', enabled: false })
+		expect(rows.writes).toEqual([])
+	})
+
 	it('saves a change the draft comparison ignores', async () => {
 		type Sched = { path: string; permissioned_as?: string }
 		const rows = fakeRows()
@@ -777,6 +808,29 @@ describe('item store: one entry per key', () => {
 		expect(rows.writes.at(-1)).toEqual({ path: 'u/me/b', value: null })
 	})

+	it('shows the draft of an editor that closed before its row was sent', async () => {
+		const rows = fakeRows()
+		const store = createItemStore(rows.port)
+		const a = adapter(async () => ({
+			deployed: deployedRes,
+			draft: rows.sent.get('u/me/r') as Res | undefined
+		}))
+		const key: ItemKey = { workspace: 'w', kind: 'resource', path: 'u/me/r' }
+		const first = store.acquire(key, { workspace: 'w', path: 'u/me/r' }, a)
+		await settle()
+		first.handle.value = { ...deployedRes, description: 'typed then closed' }
+		// The row is queued, not sent: the syncer debounces it.
+		expect(rows.sent.has('u/me/r')).toBe(false)
+		first.release()
+
+		const second = store.acquire(key, { workspace: 'w', path: 'u/me/r' }, a)
+		await settle()
+
+		// Reopening must not read past the row the closing editor left queued.
+		expect(second.handle.value?.description).toBe('typed then closed')
+		expect(second.handle.dirty).toBe(true)
+	})
+
 	it('keeps an edit typed while the row it re-reads behind is still going', async () => {
 		const rows = fakeRows()
 		const store = createItemStore(rows.port)
@@ -957,6 +1011,39 @@ describe('item store: conflicts', () => {
 		})
 	})

+	it('clears the conflict on a reload the user typed during', async () => {
+		const rows = fakeRows()
+		const read = deferred<ItemLoad<Res>>()
+		let reads = 0
+		const { item } = await open(
+			rows,
+			adapter(() => {
+				reads++
+				return reads > 1
+					? read.promise
+					: Promise.resolve({ deployed: deployedRes, draftSavedAt: 'T1' })
+			})
+		)
+		item.value = { ...deployedRes, description: 'mine' }
+		rows.conflicts.add('u/me/r')
+		expect(item.status).toBe('conflicted')
+
+		const resolving = item.resolveConflict('reload')
+		await settle()
+		// A conflicted key writes no row, so nothing typed here gets ahead of the read.
+		item.value = { ...deployedRes, description: 'typed during the reload' }
+		read.resolve({ deployed: deployedRes, draftSavedAt: 'T2' })
+		await resolving
+
+		expect(item.status).not.toBe('conflicted')
+		expect(rows.seeds.at(-1)).toEqual({ path: 'u/me/r', at: 'T2' })
+		expect(item.value?.description).toBe('typed during the reload')
+		expect(rows.writes.at(-1)).toEqual({
+			path: 'u/me/r',
+			value: { ...deployedRes, description: 'typed during the reload' }
+		})
+	})
+
 	it('drops the rejected payload when the other version is taken', async () => {
 		const rows = fakeRows()
 		const theirs = { ...deployedRes, description: 'theirs' }