diff --git a/frontend/src/lib/itemStore.svelte.ts b/frontend/src/lib/itemStore.svelte.ts index ea816450fd..6a69eeb34f 100644 --- a/frontend/src/lib/itemStore.svelte.ts +++ b/frontend/src/lib/itemStore.svelte.ts @@ -138,6 +138,8 @@ function errorMessage(e: unknown): string { * item's revision for another's. */ let nextRevision = 1 +const superseded = { ok: false, error: 'Another save of this item replaced this one' } as const + class Entry { key: ItemKey = $state()! value: V | undefined = $state() @@ -175,8 +177,11 @@ class Entry { absorbs: ((next: V, deployed: V) => boolean) | undefined refs = 0 disposed = false - /** Replaced by an entry that moved onto its key: it no longer owns the row there. */ + /** Replaced by an entry that moved onto its key: it no longer owns the row there, and a write + * reaching its turn does nothing, as its handles show the item that replaced it. */ retired = false + /** The key a save of this entry is waiting to move onto, while it waits its turn there. */ + awaiting: ItemKey | undefined handles = new Set>() stopWatch: (() => void) | undefined /** Fields a `patch` has put on the deployed side ahead of the server; a save landing @@ -258,6 +263,16 @@ class Entry { return result } + /** Let a write from another entry go next: `turn` resolves once the commands queued so far have + * run, and any queued after wait for `release`. */ + hold(): { turn: Promise; release: () => void } { + const turn = this.queue + let release!: () => void + const released = new Promise((r) => (release = r)) + this.queue = turn.then(() => released) + return { turn, release } + } + /** Count a command as started now, ahead of its turn in the queue; returns its release. */ begin(): () => void { this.commands++ @@ -354,27 +369,34 @@ class Entry { return { ok: true, path: from.path, moved: false } } const to = (adapter.pathOf ?? ((v: V) => (v as { path: string }).path))(sent) - let held: V - try { - if (!adapter.write) throw new Error('This item cannot be saved') - held = - (await adapter.write({ - ...from, - value: sent, - deployed: this.origin === 'deployed' ? snapshot(this.deployed) : undefined, - meta: this.meta - })) ?? sent - } catch (e) { - this.error = errorMessage(e) - return { ok: false, error: this.error } - } - this.error = undefined - if (held !== sent) this.adopt(sent, held) - this.deployed = { ...held, ...this.patched } as V - this.origin = 'deployed' - this.template = undefined const moved = to !== from.path - if (moved) this.moveTo(to) + const claim = moved ? this.store.claim({ ...from, path: to }, this) : undefined + try { + await claim?.turn + if (this.retired) return superseded + let held: V + try { + if (!adapter.write) throw new Error('This item cannot be saved') + held = + (await adapter.write({ + ...from, + value: sent, + deployed: this.origin === 'deployed' ? snapshot(this.deployed) : undefined, + meta: this.meta + })) ?? sent + } catch (e) { + this.error = errorMessage(e) + return { ok: false, error: this.error } + } + this.error = undefined + if (held !== sent) this.adopt(sent, held) + this.deployed = { ...held, ...this.patched } as V + this.origin = 'deployed' + this.template = undefined + if (moved) this.moveTo(to) + } finally { + claim?.release() + } this.reconcile() await this.settleRows(moved ? [from, this.key] : [this.key]) return { ok: true, path: to, moved } @@ -403,7 +425,7 @@ class Entry { discard(): Promise { return this.run(async () => { - if (!this.loaded) return { removed: false } + if (!this.loaded || this.retired) return { removed: false } if (this.origin === 'draft') { const kept = snapshot(this.value) const keptRow = this.row @@ -481,6 +503,7 @@ class Entry { } } return this.run(async () => { + if (this.retired) return superseded try { await write(this.key) } catch (e) { @@ -548,6 +571,7 @@ class Entry { return this.load(adapter) } return this.run(async () => { + if (this.retired) return superseded const desired = this.dirty ? snapshot(this.value) : null await this.ports.overwrite(this.key, desired) this.row = serialize(desired ?? undefined) ?? null @@ -560,6 +584,10 @@ class Entry { type StoreInternals = { release(entry: Entry): void rekey(entry: Entry, from: ItemKey): void + claim( + key: ItemKey, + claimant: Entry + ): { turn: Promise; release: () => void } | undefined } export type ItemHandle = { @@ -717,9 +745,38 @@ export function createItemStore(ports: ItemRowPort) { const displaced = entries.get(to) if (displaced && displaced !== entry) retire(displaced, entry) entries.set(to, entry) + }, + /** + * A save about to write the item at `key` and move onto it, while another live entry holds + * that key: it goes after the commands already queued there, and holds back any queued + * meanwhile, so one write to the item lands at a time and the displaced entry is idle when + * retired. Not when the holder is itself waiting to move onto the claimant: each would wait + * for the other. + */ + claim(key, claimant) { + const holder = entries.get(keyString(key)) + if (!holder || holder === claimant || waitsFor(holder, claimant)) return undefined + const { turn, release } = holder.hold() + claimant.awaiting = key + return { + turn: turn.then(() => (claimant.awaiting = undefined)), + release + } } } + function waitsFor(from: Entry, target: Entry): boolean { + const seen = new Set>() + for (let e = from; e.awaiting && !seen.has(e); ) { + seen.add(e) + const next = entries.get(keyString(e.awaiting)) + if (!next) return false + if (next === target) return true + e = next + } + return false + } + /** * An entry moved onto a key another live entry holds. One key has one row writer, so the * old one stops writing and every handle on it moves to the entry that now is the item — diff --git a/frontend/src/lib/itemStore.test.ts b/frontend/src/lib/itemStore.test.ts index 733350210c..8870f840b9 100644 --- a/frontend/src/lib/itemStore.test.ts +++ b/frontend/src/lib/itemStore.test.ts @@ -499,6 +499,46 @@ describe('item store: one entry per key', () => { } ]) }) + + it('writes an item it moves onto after the saves queued there, and supersedes later ones', async () => { + const rows = fakeRows() + const store = createItemStore(rows.port) + const order: string[] = [] + const gate = deferred() + const b = { ...deployedRes, path: 'u/me/b' } + const { handle: other } = store.acquire( + { workspace: 'w', kind: 'resource', path: 'u/me/b' }, + { workspace: 'w', path: 'u/me/b' }, + adapter({ deployed: b }, async (ctx) => { + await gate.promise + order.push(ctx.value.description) + }) + ) + const temporary = newItemPath() + const { handle: moving } = store.acquire( + { workspace: 'w', kind: 'resource', path: temporary }, + { workspace: 'w', path: temporary, template: b }, + adapter({}, async (ctx) => void order.push(ctx.value.description)) + ) + await settle() + + other.value = { ...b, description: 'first' } + const first = other.save() + moving.value = { ...b, description: 'second' } + const second = moving.save() + await settle() + other.value = { ...b, description: 'queued behind the move' } + const third = other.save() + await settle() + expect(order).toEqual([]) + + gate.resolve() + expect(await first).toMatchObject({ ok: true }) + expect(await second).toMatchObject({ ok: true, moved: true }) + expect(await third).toMatchObject({ ok: false }) + expect(order).toEqual(['first', 'second']) + expect(other.deployed?.description).toBe('second') + }) }) describe('item store: conflicts', () => {