Files
windmill/frontend/src/lib/userScopedDb.ts
T
796b6e5297 feat: back AI sessions up to the workspace object storage (#11116)
* feat: back AI sessions up to the workspace object storage

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SG5qEPM6Fmf7VerXS5nnWp

* fix: bind the backup key to the user and pack pushes within the server caps

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SG5qEPM6Fmf7VerXS5nnWp

* fix: keep refused and unavailable marks, one mark per key, stream the flush

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SG5qEPM6Fmf7VerXS5nnWp

* fix: settle only fully sent sessions, keep removals while backups are off, cap pull bodies

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01SG5qEPM6Fmf7VerXS5nnWp

* fix: bound removal marks while backups are off and stale the sync rows instead of dropping them

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: retry a lost lock, cap nested push lists and oversized pieces, drop a stale copy of a chat that outgrew the backup

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: cap pieces per push, size requests in UTF-8, keep a move's removal for an off workspace, disclose the restore counter

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: file a move's removal only once the new copy landed, retire marks through the sync row

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: keep a session marked while deletes are carried over, drop only gone sessions' marks

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: record what a refused flush already stored, stop early when every mark is retired

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: restore past another workspace's removal mark, file a move's removal before its row, bound the first pulled session

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: re-key the backups on workspace key rotation, accept only base64 images, carry a delete on the sync row when its mark cannot be written

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: durable conditional re-key of session backups, re-push on a storage switch

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: fail a push the key rotated under, settle no session split across storages, narrow the re-key module

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: keep a delete filed during a push, bound the pull and re-key listings

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: bound the session listing, mark the store's own user on a write that lands after a user switch

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: list sessions through per-session index markers, hold a session's parts back after a failed one

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: record a rotation on every build, list a session only on the part that completes its push

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: leave an object larger than any push writes unread

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: read each object against its listed size, carry a dirty mark that cannot be written on the sync row

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: note the storages the re-key walk completed on, reach another user's rows on a failed mark, read a head at its cap

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: read the replaced key under its row lock, carry a refused dirty bump on the sync row

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: pull a session that outgrew one answer in pages, imported only whole

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: build a pull page from the smallest keys of the whole listing, stage each page as it lands

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: re-record a key rotated back to, admit earlier-page images, restage over a cut-short restore

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* refactor: delete the backups on key rotation instead of re-keying them

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: end a pull page before an object that grew since the listing, prune what a cut-short restore staged

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: delete the backups before the key commits, skip a planted object whatever its listing says, lock a restore across tabs, prune stale artifact versions

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: keep the backups under a prefix named by the key, delete the previous key's prefix after the commit

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: name the backup prefix by a generation the rotation bumps, never write an older record over a newer one on restore

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: retire a removal only against the storage holding the backup, restart a paged pull whose listing moved

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: fingerprint a pull page before reading it, answer the backup generation apart from the storage identity

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: answer needs_head for a headless session push, prune restaged pieces by id

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: serialize a session's push and removal, open whole pushes with the head, prune only own restores

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: require a head on a whole push, prune before the record lands, restore only under Web Locks

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: incremental pushes ride on a listed session, removals wait for every storage holding a copy

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: a whole push replaces the backup under a per-push token, a pull page is checked after its reads

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: fingerprint a pull page by entity tag and version too

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: a moved session's removal mark names the storages holding the old copy

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: restore a workspace family together, the newest copy of a moved session winning

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: name the marker by the session's move count, abort a family restore a listing failed in

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: list the family again before a restored record lands, require the pull fingerprint

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: list the whole family once per restored workspace, off members included

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: a push split over parts, incremental too, unlists the session until its last part

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: refuse a partial push part that names no push

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: keep a refused bump for a session with no row yet, probe an off workspace again

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: backfill row-carried bumps after a reload, ask an off workspace again on a timer

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: backfill a row for its bumps only when it carries some

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* fix: a backfilled mark that cannot be written counts from the page's counter

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* docs: say an off workspace is asked again, in the mirror's comments

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* chore: update ee-repo-ref to 1c1dab33563c4907aff8b0da825fb66db60af82a

This commit updates the EE repository reference after PR #796 was merged in windmill-ee-private.

Previous ee-repo-ref: 289b477ca3fc993da06ec09b11c8f55d5e4e39c1

New ee-repo-ref: 1c1dab33563c4907aff8b0da825fb66db60af82a

Automated by sync-ee-ref workflow.

* fix: unlist a session while an incremental push changes more than one object

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
2026-09-15 15:48:33 +00:00

243 lines
9.4 KiB
TypeScript

import {
openDB as idbOpenDB,
deleteDB as idbDeleteDB,
type DBSchema,
type IDBPDatabase,
type IDBPTransaction,
type StoreNames
} from 'idb'
import { scopedKey } from '$lib/userScopedStorage'
// Per-user IndexedDB lifecycle, shared by the session list and the copilot
// chat-history stores. The effective DB name is the base name namespaced by the
// logged-in user's email (scopedKey), so two users on a shared browser never
// touch the same physical database.
//
// `whenReady()` is name-aware: it computes the current scoped name on every call
// and transparently closes + reopens when the email changes, so the handle
// self-heals on user switch WITHOUT subscribing to onUserChange. That matters
// because there is one handle per HistoryManager instance (singleton + one per
// session runtime) — a per-instance subscription would leak callbacks. The email
// can change back and forth within one page: logout is a client-side goto() that
// clears userStore, not a reload.
export interface UserScopedDbMigrateDeps {
openDB: typeof idbOpenDB
deleteDB: typeof idbDeleteDB
}
export interface UserScopedDbOptions<Schema extends DBSchema> {
version: number
// The version-change transaction is the only way to add an index to a store that
// already exists; a store being created gets it from the store handle instead.
upgrade: (
db: IDBPDatabase<Schema>,
tx: IDBPTransaction<Schema, StoreNames<Schema>[], 'versionchange'>
) => void
// Invoked once per scoped name right after a successful open. The fn owns its
// own "already migrated / not applicable" gate (e.g. checking a store's
// count) — claim-then-delete legacy data lives here.
migrate?: (db: IDBPDatabase<Schema>, deps: UserScopedDbMigrateDeps) => Promise<void>
// Injectable for tests (defaults to the real idb implementations).
openDB?: typeof idbOpenDB
deleteDB?: typeof idbDeleteDB
// How long an open waits to settle before giving up. Injectable for tests, which
// cannot afford the real grace period.
openGraceMs?: number
}
export interface UserScopedDb<Schema extends DBSchema> {
// Resolves to the open DB for the current user, or undefined when no user is logged
// in yet, the open failed, or another connection is holding up a schema upgrade —
// directly, or by sitting ahead of this open in the browser's queue for the database
// (degrade to in-memory; never rejects, never hangs).
whenReady(): Promise<IDBPDatabase<Schema> | undefined>
close(): void
}
// How long an open waits to settle before the opener gives up and degrades to in-memory.
// Only a connection that ignores `versionchange`, or an open queued behind one, takes this long.
const OPEN_GRACE_MS = 5000
export function userScopedDb<Schema extends DBSchema>(
baseName: string,
opts: UserScopedDbOptions<Schema>
): UserScopedDb<Schema> {
const openDB = opts.openDB ?? idbOpenDB
const deleteDB = opts.deleteDB ?? idbDeleteDB
const openGraceMs = opts.openGraceMs ?? OPEN_GRACE_MS
const migratedNames = new Set<string>()
/**
* One open request for one scoped database, plus everything learned about it since.
*
* An open cannot be cancelled and the browser processes a database's opens in order, so
* a replacement issued while one is pending waits behind it, never reaching its own
* `blocked` callback. An attempt is therefore replaced only once it has settled; losing
* interest in it is recorded on these fields instead.
*/
interface Attempt {
/**
* The open with the keep-or-discard decision already applied, so a handle we drop is
* reported as undefined rather than handed over and closed behind the caller's back.
*/
outcome: Promise<IDBPDatabase<Schema> | undefined>
/** Set when `outcome` settles. Its presence, not its `db`, is what "settled" means. */
settled?: { db: IDBPDatabase<Schema> | undefined }
/** Resolves undefined when the open has taken too long to settle. */
gaveUp: Promise<undefined>
timedOut: boolean
/** Handle yielded to another tab's upgrade, or force-closed by the browser. */
dead: boolean
/** Released before it arrived: close it on arrival rather than hand it out. */
unwanted: boolean
}
const attempts = new Map<string, Attempt>()
let currentName: string | undefined
async function open(
name: string,
attempt: Attempt,
onTooLong: () => void
): Promise<IDBPDatabase<Schema> | undefined> {
// Armed before the request is issued rather than from `blocked`, because an open
// queued behind another one is told nothing at all: it waits its turn in the
// browser's per-database queue, and a callback that never fires cannot bound it.
let graceTimer: ReturnType<typeof setTimeout> | undefined = setTimeout(onTooLong, openGraceMs)
const stopWaiting = () => {
if (graceTimer) clearTimeout(graceTimer)
graceTimer = undefined
}
try {
let handle: IDBPDatabase<Schema> | undefined
const db = await openDB<Schema>(name, opts.version, {
upgrade(database, _oldVersion, _newVersion, transaction) {
// The version-change transaction is ours: nothing is queued ahead of this
// open any more, and what remains is our own upgrade running.
stopWaiting()
opts.upgrade(database, transaction)
},
// Another tab is opening this database at a higher version, which our open
// connection would block indefinitely. Let go so their upgrade lands; this
// attempt is replaced by a fresh open once it has settled.
blocking() {
handle?.close()
attempt.dead = true
},
blocked(currentVersion, blockedVersion) {
console.warn(
`userScopedDb(${baseName}): upgrade ${currentVersion}→${blockedVersion} waiting on another connection`
)
},
// The browser force-closed the connection (site data cleared, database
// dropped from devtools). Every request on this handle would now throw.
terminated() {
attempt.dead = true
}
})
handle = db
// The handle is in hand; a slow migrate is not a hung open.
stopWaiting()
if (opts.migrate && !migratedNames.has(name)) {
migratedNames.add(name)
try {
await opts.migrate(db, { openDB, deleteDB })
} catch (e) {
// A failed migration is non-fatal: the (open) DB is still usable, so
// we log and return it — unlike a failed open below, which yields
// undefined. Worst case the legacy claim is missed, not the store.
console.error(`userScopedDb(${baseName}): migration failed`, e)
}
}
return db
} catch (e) {
// Failed open (corrupt / private-browsing / a newer version already stored):
// degrade to in-memory by resolving undefined (callers no-op their writes).
console.error(`userScopedDb(${baseName}): could not open database`, e)
return undefined
} finally {
stopWaiting()
}
}
function start(name: string): Attempt {
let giveUp!: () => void
const attempt: Attempt = {
// Assigned immediately below; open() needs the attempt to report back onto.
outcome: undefined as unknown as Promise<IDBPDatabase<Schema> | undefined>,
gaveUp: new Promise<undefined>((resolve) => (giveUp = () => resolve(undefined))),
timedOut: false,
dead: false,
unwanted: false
}
// The timer is owned by this attempt, so it can only ever time out its own request.
attempt.outcome = open(name, attempt, () => {
attempt.timedOut = true
giveUp()
}).then((db) => {
// Arrived after we stopped wanting it, or after we yielded the handle: holding it
// open would block the next tab's upgrade for no one's benefit, and by now it may
// belong to a user who is no longer logged in.
const discard = attempt.unwanted || attempt.dead
if (discard) {
db?.close()
attempt.dead = true
}
attempt.settled = { db: discard ? undefined : db }
attempt.timedOut = false
return attempt.settled.db
})
attempts.set(name, attempt)
return attempt
}
/** Stop serving `name`; a still-pending attempt is only marked, per the rule on Attempt. */
function release(name: string | undefined) {
const attempt = name ? attempts.get(name) : undefined
if (!attempt) return
if (attempt.settled) {
const { db } = attempt.settled
attempts.delete(name!)
// A microtask later, never synchronously: callers hold this handle across several
// awaits (hydrate, write-behind), and closing under them turns a routine user
// switch into InvalidStateError. close() then waits on what they started.
if (db) void Promise.resolve().then(() => db.close())
return
}
attempt.unwanted = true
}
return {
whenReady() {
const name = scopedKey(baseName)
if (name !== currentName) {
release(currentName)
currentName = name
}
if (!name) return Promise.resolve(undefined)
let attempt = attempts.get(name)
// The one place an attempt is replaced, and only once settled — see Attempt.
if (attempt?.settled && attempt.dead) {
attempts.delete(name)
attempt = undefined
}
attempt ??= start(name)
// Wanting it again cancels a release that has not landed yet.
attempt.unwanted = false
if (attempt.settled) return Promise.resolve(attempt.settled.db)
// Still in flight. `dead` means the handle is spoken for and `timedOut` that we
// stopped waiting — either way, degrade now rather than block the caller; both
// resolve themselves when the request finally settles.
if (attempt.dead || attempt.timedOut) return Promise.resolve(undefined)
return Promise.race([attempt.outcome, attempt.gaveUp])
},
close() {
release(currentName)
currentName = undefined
}
}
}