mirror of
https://github.com/hellodigua/ChatLab.git
synced 2026-10-26 02:04:37 +08:00
fix(web): support multi-tab workspace access
This commit is contained in:
1 parent
d0a9a03f7c
commit
ce603bdc52
26 files changed
+834
-269
No files matched your search
@@ -1,5 +1,5 @@
|
||||
<script setup lang="ts">
|
||||
import { computed, onBeforeUnmount, onMounted, ref, watch } from 'vue'
|
||||
import { computed, onMounted, ref, watch } from 'vue'
|
||||
import { useColorMode, useMediaQuery } from '@vueuse/core'
|
||||
import { storeToRefs } from 'pinia'
|
||||
import { useRoute } from 'vue-router'
|
||||
@@ -16,13 +16,6 @@ import { useSessionStore } from '@/stores/session'
|
||||
import { useSettingsStore } from '@/stores/settings'
|
||||
import { StartupLoading } from '@/components/UI'
|
||||
import { PLATFORM_CAPABILITIES } from '@/utils/platform-capabilities'
|
||||
import {
|
||||
WebWasmWorkspaceBusyError,
|
||||
acquireWebWasmWorkspaceLease,
|
||||
handleWebWasmWorkspacePageHide,
|
||||
type WebWasmLockManager,
|
||||
type WebWasmWorkspaceLease,
|
||||
} from './workspace-lock'
|
||||
|
||||
const { t } = useI18n()
|
||||
const route = useRoute()
|
||||
@@ -57,7 +50,6 @@ const toaster = {
|
||||
}
|
||||
|
||||
let initInProgress = false
|
||||
let workspaceLease: WebWasmWorkspaceLease | undefined
|
||||
|
||||
async function initializeApp() {
|
||||
if (initInProgress || isInitialized.value) return
|
||||
@@ -65,7 +57,6 @@ async function initializeApp() {
|
||||
initError.value = null
|
||||
|
||||
try {
|
||||
workspaceLease ??= await acquireWebWasmWorkspaceLease(navigator.locks as WebWasmLockManager | undefined)
|
||||
await initializeAppRuntime({
|
||||
capabilities: PLATFORM_CAPABILITIES,
|
||||
initializeServices: () => initServices(),
|
||||
@@ -77,7 +68,7 @@ async function initializeApp() {
|
||||
} catch (error) {
|
||||
console.error('Web WASM application initialization failed', error)
|
||||
initError.value =
|
||||
error instanceof WebWasmWorkspaceBusyError
|
||||
error instanceof Error && 'code' in error && error.code === 'OPFS_WORKSPACE_BUSY'
|
||||
? t('common.webWasmWorkspaceBusy')
|
||||
: error instanceof Error
|
||||
? error.message
|
||||
@@ -88,23 +79,9 @@ async function initializeApp() {
|
||||
}
|
||||
}
|
||||
|
||||
function releaseWorkspaceLease() {
|
||||
workspaceLease?.release()
|
||||
workspaceLease = undefined
|
||||
}
|
||||
|
||||
function handlePageHide(event: PageTransitionEvent) {
|
||||
workspaceLease = handleWebWasmWorkspacePageHide(event, workspaceLease)
|
||||
}
|
||||
|
||||
onMounted(() => {
|
||||
window.addEventListener('pagehide', handlePageHide)
|
||||
void initializeApp()
|
||||
})
|
||||
onBeforeUnmount(() => {
|
||||
window.removeEventListener('pagehide', handlePageHide)
|
||||
releaseWorkspaceLease()
|
||||
})
|
||||
</script>
|
||||
|
||||
<template>
|
||||
|
||||
@@ -14,15 +14,23 @@ import { installGlobalErrorReporting, reportError, reportRuntimeLog } from '@/se
|
||||
import { initServices } from '@/services/registry'
|
||||
import { registerWebWasmAdapters } from '@/services/browser-runtime/register'
|
||||
import { handleWebWasmPageHide } from './runtime-lifecycle'
|
||||
import { WebWasmSessionSync } from './session-sync'
|
||||
import { useSessionStore } from '@/stores/session'
|
||||
import '@/assets/styles/main.css'
|
||||
|
||||
async function start(): Promise<void> {
|
||||
reportRuntimeLog({ level: 'info', scope: 'web-wasm-bootstrap', message: 'Initializing Web WASM services' })
|
||||
installGlobalErrorReporting()
|
||||
await initServices({ initializeWebWasm: registerWebWasmAdapters })
|
||||
const sessionSync = new WebWasmSessionSync()
|
||||
await initServices({
|
||||
initializeWebWasm: (registry) =>
|
||||
registerWebWasmAdapters({
|
||||
...registry,
|
||||
onWorkspaceChanged: (event) => sessionSync.publish(event),
|
||||
}),
|
||||
})
|
||||
|
||||
const runtime = useBrowserRuntimeService()
|
||||
window.addEventListener('pagehide', (event) => handleWebWasmPageHide(event, () => runtime.dispose()))
|
||||
|
||||
const app = createApp(App)
|
||||
app.config.errorHandler = (error, _instance, info) => {
|
||||
@@ -39,6 +47,35 @@ async function start(): Promise<void> {
|
||||
app.use(router)
|
||||
app.use(ui)
|
||||
app.use(i18n)
|
||||
|
||||
const sessionStore = useSessionStore(pinia)
|
||||
let refreshRequested = false
|
||||
let refreshInProgress = false
|
||||
const refreshSessions = async () => {
|
||||
refreshRequested = true
|
||||
if (refreshInProgress) return
|
||||
refreshInProgress = true
|
||||
try {
|
||||
// 连续收到多个标签页事件时至少执行到最后一次,避免导入/重命名紧邻发生时漏掉最新状态。
|
||||
while (refreshRequested) {
|
||||
refreshRequested = false
|
||||
await sessionStore.loadSessions()
|
||||
}
|
||||
} finally {
|
||||
refreshInProgress = false
|
||||
}
|
||||
}
|
||||
const unsubscribeSessionSync = sessionSync.subscribe(() => {
|
||||
void refreshSessions()
|
||||
})
|
||||
|
||||
window.addEventListener('pagehide', (event) =>
|
||||
handleWebWasmPageHide(event, () => {
|
||||
unsubscribeSessionSync()
|
||||
sessionSync.dispose()
|
||||
runtime.dispose()
|
||||
})
|
||||
)
|
||||
app.mount('#app')
|
||||
}
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
<script setup lang="ts">
|
||||
import { computed } from 'vue'
|
||||
import { computed, watch } from 'vue'
|
||||
import { storeToRefs } from 'pinia'
|
||||
import { useI18n } from 'vue-i18n'
|
||||
import { useRoute, useRouter } from 'vue-router'
|
||||
@@ -13,7 +13,7 @@ const { t } = useI18n()
|
||||
const route = useRoute()
|
||||
const router = useRouter()
|
||||
const sessionStore = useSessionStore()
|
||||
const { currentSessionId } = storeToRefs(sessionStore)
|
||||
const { currentSessionId, isInitialized, sessions } = storeToRefs(sessionStore)
|
||||
|
||||
const tabs = [{ id: 'insights', labelKey: 'analysis.tabs.insights', icon: 'i-heroicons-presentation-chart-bar' }]
|
||||
|
||||
@@ -57,6 +57,20 @@ const otherMemberAvatar = computed(() => {
|
||||
const loadErrorText = computed(() =>
|
||||
t(route.name === 'private-chat' ? 'analysis.privateChat.loadError' : 'analysis.groupChat.loadError')
|
||||
)
|
||||
|
||||
watch(
|
||||
() => sessions.value.find((item) => item.id === route.params.id),
|
||||
(catalogSession) => {
|
||||
if (!isInitialized.value) return
|
||||
if (!catalogSession) {
|
||||
session.value = null
|
||||
void router.replace('/')
|
||||
return
|
||||
}
|
||||
// 其他标签页重命名后,同步更新当前详情页标题,不必重新执行整组分析查询。
|
||||
if (session.value) session.value = catalogSession
|
||||
}
|
||||
)
|
||||
</script>
|
||||
|
||||
<template>
|
||||
|
||||
@@ -0,0 +1,77 @@
|
||||
import assert from 'node:assert/strict'
|
||||
import { describe, it } from 'node:test'
|
||||
import { WebWasmSessionSync, type WebWasmSessionSyncChannel } from './session-sync'
|
||||
|
||||
class FakeChannelBus {
|
||||
readonly channels = new Set<FakeChannel>()
|
||||
|
||||
create(): FakeChannel {
|
||||
const channel = new FakeChannel(this)
|
||||
this.channels.add(channel)
|
||||
return channel
|
||||
}
|
||||
}
|
||||
|
||||
class FakeChannel implements WebWasmSessionSyncChannel {
|
||||
private readonly listeners = new Set<(event: MessageEvent<unknown>) => void>()
|
||||
|
||||
constructor(private readonly bus: FakeChannelBus) {}
|
||||
|
||||
postMessage(message: unknown): void {
|
||||
for (const channel of this.bus.channels) {
|
||||
if (channel === this) continue
|
||||
channel.emit(message)
|
||||
}
|
||||
}
|
||||
|
||||
addEventListener(_type: 'message', listener: (event: MessageEvent<unknown>) => void): void {
|
||||
this.listeners.add(listener)
|
||||
}
|
||||
|
||||
removeEventListener(_type: 'message', listener: (event: MessageEvent<unknown>) => void): void {
|
||||
this.listeners.delete(listener)
|
||||
}
|
||||
|
||||
close(): void {
|
||||
this.bus.channels.delete(this)
|
||||
this.listeners.clear()
|
||||
}
|
||||
|
||||
private emit(message: unknown): void {
|
||||
for (const listener of this.listeners) listener({ data: message } as MessageEvent<unknown>)
|
||||
}
|
||||
}
|
||||
|
||||
describe('WebWasmSessionSync', () => {
|
||||
it('notifies other tabs about successful session mutations without echoing to the sender', () => {
|
||||
const bus = new FakeChannelBus()
|
||||
const first = new WebWasmSessionSync(bus.create())
|
||||
const second = new WebWasmSessionSync(bus.create())
|
||||
const firstEvents: string[] = []
|
||||
const secondEvents: string[] = []
|
||||
first.subscribe((event) => firstEvents.push(`${event.type}:${event.sessionId}`))
|
||||
second.subscribe((event) => secondEvents.push(`${event.type}:${event.sessionId}`))
|
||||
|
||||
first.publish({ type: 'rename', sessionId: 'session-one' })
|
||||
|
||||
assert.deepEqual(firstEvents, [])
|
||||
assert.deepEqual(secondEvents, ['rename:session-one'])
|
||||
first.dispose()
|
||||
second.dispose()
|
||||
})
|
||||
|
||||
it('ignores malformed channel messages', () => {
|
||||
const bus = new FakeChannelBus()
|
||||
const sender = bus.create()
|
||||
const receiver = new WebWasmSessionSync(bus.create())
|
||||
let received = false
|
||||
receiver.subscribe(() => {
|
||||
received = true
|
||||
})
|
||||
|
||||
sender.postMessage({ type: 'unknown', sessionId: 'session-one' })
|
||||
|
||||
assert.equal(received, false)
|
||||
receiver.dispose()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,51 @@
|
||||
import type { WebRuntimeWorkspaceChangeEvent } from '@openchatlab/web-runtime'
|
||||
|
||||
export const WEB_WASM_SESSION_SYNC_CHANNEL = 'chatlab-web-wasm-session-sync'
|
||||
|
||||
export interface WebWasmSessionSyncChannel {
|
||||
postMessage(message: unknown): void
|
||||
addEventListener(type: 'message', listener: (event: MessageEvent<unknown>) => void): void
|
||||
removeEventListener(type: 'message', listener: (event: MessageEvent<unknown>) => void): void
|
||||
close(): void
|
||||
}
|
||||
|
||||
export class WebWasmSessionSync {
|
||||
private readonly listeners = new Set<(event: WebRuntimeWorkspaceChangeEvent) => void>()
|
||||
|
||||
constructor(private readonly channel: WebWasmSessionSyncChannel | undefined = createChannel()) {
|
||||
this.channel?.addEventListener('message', this.handleMessage)
|
||||
}
|
||||
|
||||
publish(event: WebRuntimeWorkspaceChangeEvent): void {
|
||||
this.channel?.postMessage(event)
|
||||
}
|
||||
|
||||
subscribe(listener: (event: WebRuntimeWorkspaceChangeEvent) => void): () => void {
|
||||
this.listeners.add(listener)
|
||||
return () => this.listeners.delete(listener)
|
||||
}
|
||||
|
||||
dispose(): void {
|
||||
this.channel?.removeEventListener('message', this.handleMessage)
|
||||
this.channel?.close()
|
||||
this.listeners.clear()
|
||||
}
|
||||
|
||||
private readonly handleMessage = (event: MessageEvent<unknown>): void => {
|
||||
if (!isWorkspaceChangeEvent(event.data)) return
|
||||
for (const listener of this.listeners) listener(event.data)
|
||||
}
|
||||
}
|
||||
|
||||
function createChannel(): WebWasmSessionSyncChannel | undefined {
|
||||
return typeof BroadcastChannel === 'function' ? new BroadcastChannel(WEB_WASM_SESSION_SYNC_CHANNEL) : undefined
|
||||
}
|
||||
|
||||
function isWorkspaceChangeEvent(value: unknown): value is WebRuntimeWorkspaceChangeEvent {
|
||||
if (!value || typeof value !== 'object') return false
|
||||
const event = value as Partial<WebRuntimeWorkspaceChangeEvent>
|
||||
return (
|
||||
(event.type === 'import' || event.type === 'delete' || event.type === 'rename') &&
|
||||
typeof event.sessionId === 'string'
|
||||
)
|
||||
}
|
||||
@@ -1,71 +0,0 @@
|
||||
import assert from 'node:assert/strict'
|
||||
import { describe, it } from 'node:test'
|
||||
import {
|
||||
WebWasmWorkspaceBusyError,
|
||||
acquireWebWasmWorkspaceLease,
|
||||
handleWebWasmWorkspacePageHide,
|
||||
type WebWasmLockManager,
|
||||
type WebWasmWorkspaceLease,
|
||||
} from './workspace-lock'
|
||||
|
||||
describe('acquireWebWasmWorkspaceLease', () => {
|
||||
it('rejects a second Web WASM page while the workspace lock is occupied', async () => {
|
||||
const lockManager: WebWasmLockManager = {
|
||||
async request(_name, options, callback) {
|
||||
assert.deepEqual(options, { mode: 'exclusive', ifAvailable: true })
|
||||
await callback(null)
|
||||
},
|
||||
}
|
||||
|
||||
await assert.rejects(acquireWebWasmWorkspaceLease(lockManager), WebWasmWorkspaceBusyError)
|
||||
})
|
||||
|
||||
it('holds the workspace lock until the lease is released', async () => {
|
||||
let requestCompleted = false
|
||||
const lockManager: WebWasmLockManager = {
|
||||
async request(_name, _options, callback) {
|
||||
await callback({})
|
||||
requestCompleted = true
|
||||
},
|
||||
}
|
||||
|
||||
const lease = await acquireWebWasmWorkspaceLease(lockManager)
|
||||
assert.equal(requestCompleted, false)
|
||||
|
||||
lease.release()
|
||||
await new Promise((resolve) => setTimeout(resolve, 0))
|
||||
assert.equal(requestCompleted, true)
|
||||
|
||||
lease.release()
|
||||
})
|
||||
})
|
||||
|
||||
describe('handleWebWasmWorkspacePageHide', () => {
|
||||
it('keeps the workspace lease while the page enters the back-forward cache', () => {
|
||||
let releaseCount = 0
|
||||
const lease: WebWasmWorkspaceLease = {
|
||||
release() {
|
||||
releaseCount++
|
||||
},
|
||||
}
|
||||
|
||||
const retainedLease = handleWebWasmWorkspacePageHide({ persisted: true }, lease)
|
||||
|
||||
assert.equal(retainedLease, lease)
|
||||
assert.equal(releaseCount, 0)
|
||||
})
|
||||
|
||||
it('releases and clears the workspace lease when the page is discarded', () => {
|
||||
let releaseCount = 0
|
||||
const lease: WebWasmWorkspaceLease = {
|
||||
release() {
|
||||
releaseCount++
|
||||
},
|
||||
}
|
||||
|
||||
const retainedLease = handleWebWasmWorkspacePageHide({ persisted: false }, lease)
|
||||
|
||||
assert.equal(retainedLease, undefined)
|
||||
assert.equal(releaseCount, 1)
|
||||
})
|
||||
})
|
||||
@@ -1,75 +0,0 @@
|
||||
import { handleWebWasmPageHide } from './runtime-lifecycle'
|
||||
|
||||
export const WEB_WASM_WORKSPACE_LOCK = 'chatlab-web-wasm-opfs-workspace'
|
||||
|
||||
export interface WebWasmLockManager {
|
||||
request(
|
||||
name: string,
|
||||
options: { mode: 'exclusive'; ifAvailable: true },
|
||||
callback: (lock: object | null) => Promise<void>
|
||||
): Promise<void>
|
||||
}
|
||||
|
||||
export interface WebWasmWorkspaceLease {
|
||||
release(): void
|
||||
}
|
||||
|
||||
export class WebWasmWorkspaceBusyError extends Error {
|
||||
readonly code = 'OPFS_WORKSPACE_BUSY'
|
||||
|
||||
constructor() {
|
||||
super('The Web WASM workspace is already open in another tab')
|
||||
this.name = 'WebWasmWorkspaceBusyError'
|
||||
}
|
||||
}
|
||||
|
||||
export function handleWebWasmWorkspacePageHide(
|
||||
event: Pick<PageTransitionEvent, 'persisted'>,
|
||||
lease: WebWasmWorkspaceLease | undefined
|
||||
): WebWasmWorkspaceLease | undefined {
|
||||
let retainedLease = lease
|
||||
handleWebWasmPageHide(event, () => {
|
||||
lease?.release()
|
||||
retainedLease = undefined
|
||||
})
|
||||
return retainedLease
|
||||
}
|
||||
|
||||
export async function acquireWebWasmWorkspaceLease(
|
||||
lockManager: WebWasmLockManager | undefined
|
||||
): Promise<WebWasmWorkspaceLease> {
|
||||
if (!lockManager) return { release: () => undefined }
|
||||
|
||||
let releaseLock: () => void = () => undefined
|
||||
const lockReleased = new Promise<void>((resolve) => {
|
||||
releaseLock = resolve
|
||||
})
|
||||
let released = false
|
||||
const lease: WebWasmWorkspaceLease = {
|
||||
release() {
|
||||
if (released) return
|
||||
released = true
|
||||
releaseLock()
|
||||
},
|
||||
}
|
||||
|
||||
let resolveAcquired!: (lease: WebWasmWorkspaceLease) => void
|
||||
let rejectAcquired!: (error: unknown) => void
|
||||
const acquired = new Promise<WebWasmWorkspaceLease>((resolve, reject) => {
|
||||
resolveAcquired = resolve
|
||||
rejectAcquired = reject
|
||||
})
|
||||
|
||||
void lockManager
|
||||
.request(WEB_WASM_WORKSPACE_LOCK, { mode: 'exclusive', ifAvailable: true }, async (lock) => {
|
||||
if (!lock) {
|
||||
rejectAcquired(new WebWasmWorkspaceBusyError())
|
||||
return
|
||||
}
|
||||
resolveAcquired(lease)
|
||||
await lockReleased
|
||||
})
|
||||
.catch(rejectAcquired)
|
||||
|
||||
return acquired
|
||||
}
|
||||
@@ -8,6 +8,7 @@ export type {
|
||||
OpenDatabaseResult,
|
||||
RpcProgressPayload,
|
||||
RuntimeLogEvent,
|
||||
WebRuntimeWorkspaceChangeEvent,
|
||||
WebRuntimeTaskMap,
|
||||
WebRuntimeTaskPayload,
|
||||
WebRuntimeTaskResult,
|
||||
|
||||
@@ -177,6 +177,90 @@ describe('WebRuntimeRpcClient', () => {
|
||||
client.dispose()
|
||||
})
|
||||
|
||||
it('reports successful workspace mutations after the Worker releases storage', async () => {
|
||||
const worker = new FakeWorker()
|
||||
const changes: Array<{ type: string; sessionId: string }> = []
|
||||
const client = new WebRuntimeRpcClient(() => worker, {
|
||||
onWorkspaceChanged: (event) => changes.push(event),
|
||||
})
|
||||
|
||||
const renamed = client.request('session.rename', { sessionId: 'session-one', name: 'Renamed' })
|
||||
const renameRequest = worker.messages[0] as { id: string }
|
||||
worker.emitMessage({
|
||||
id: renameRequest.id,
|
||||
type: 'result',
|
||||
payload: { taskType: 'session.rename', result: { renamed: true } },
|
||||
})
|
||||
await renamed
|
||||
|
||||
const missingDelete = client.request('session.delete', { sessionId: 'missing' })
|
||||
const deleteRequest = worker.messages[1] as { id: string }
|
||||
worker.emitMessage({
|
||||
id: deleteRequest.id,
|
||||
type: 'result',
|
||||
payload: { taskType: 'session.delete', result: { deleted: false } },
|
||||
})
|
||||
await missingDelete
|
||||
|
||||
const imported = client.request('import.start', {
|
||||
source: {
|
||||
name: 'fixture.json',
|
||||
size: 2,
|
||||
text: async () => '{}',
|
||||
arrayBuffer: async () => new ArrayBuffer(0),
|
||||
slice: () => new Blob(),
|
||||
},
|
||||
})
|
||||
const importRequest = worker.messages[2] as { id: string }
|
||||
worker.emitMessage({
|
||||
id: importRequest.id,
|
||||
type: 'result',
|
||||
payload: {
|
||||
taskType: 'import.start',
|
||||
result: {
|
||||
sessionId: 'session-two',
|
||||
formatId: 'chatlab',
|
||||
messageCount: 1,
|
||||
memberCount: 1,
|
||||
skippedCount: 0,
|
||||
},
|
||||
},
|
||||
})
|
||||
await imported
|
||||
|
||||
assert.deepEqual(changes, [
|
||||
{ type: 'rename', sessionId: 'session-one' },
|
||||
{ type: 'import', sessionId: 'session-two' },
|
||||
])
|
||||
client.dispose()
|
||||
})
|
||||
|
||||
it('still resolves a mutation when the workspace change callback fails', async () => {
|
||||
const worker = new FakeWorker()
|
||||
const client = new WebRuntimeRpcClient(() => worker, {
|
||||
onWorkspaceChanged: () => {
|
||||
throw new Error('channel closed')
|
||||
},
|
||||
})
|
||||
const originalError = console.error
|
||||
console.error = () => undefined
|
||||
|
||||
try {
|
||||
const renamed = client.request('session.rename', { sessionId: 'session-one', name: 'Renamed' })
|
||||
const request = worker.messages[0] as { id: string }
|
||||
worker.emitMessage({
|
||||
id: request.id,
|
||||
type: 'result',
|
||||
payload: { taskType: 'session.rename', result: { renamed: true } },
|
||||
})
|
||||
|
||||
assert.deepEqual(await renamed, { renamed: true })
|
||||
} finally {
|
||||
console.error = originalError
|
||||
client.dispose()
|
||||
}
|
||||
})
|
||||
|
||||
it('rejects pending work after a crash and creates a fresh worker for the next request', async () => {
|
||||
const workers: FakeWorker[] = []
|
||||
const client = new WebRuntimeRpcClient(() => {
|
||||
|
||||
@@ -3,6 +3,7 @@ import type {
|
||||
RpcResponseEnvelope,
|
||||
RuntimeLogEvent,
|
||||
SerializedRpcError,
|
||||
WebRuntimeWorkspaceChangeEvent,
|
||||
WebRuntimeTaskPayload,
|
||||
WebRuntimeTaskResult,
|
||||
WebRuntimeTaskType,
|
||||
@@ -24,9 +25,12 @@ export interface RpcRequestOptions {
|
||||
|
||||
export interface WebRuntimeRpcClientOptions {
|
||||
onLog?: (event: RuntimeLogEvent) => void
|
||||
onWorkspaceChanged?: (event: WebRuntimeWorkspaceChangeEvent) => void
|
||||
}
|
||||
|
||||
interface PendingRequest {
|
||||
taskType: WebRuntimeTaskType
|
||||
payload: WebRuntimeTaskPayload<WebRuntimeTaskType>
|
||||
resolve(value: unknown): void
|
||||
reject(error: Error): void
|
||||
onProgress?: (event: RpcProgressPayload) => void
|
||||
@@ -74,6 +78,8 @@ export class WebRuntimeRpcClient {
|
||||
|
||||
return new Promise<WebRuntimeTaskResult<T>>((resolve, reject) => {
|
||||
const pending: PendingRequest = {
|
||||
taskType: type,
|
||||
payload: payload as WebRuntimeTaskPayload<WebRuntimeTaskType>,
|
||||
resolve: resolve as (value: unknown) => void,
|
||||
reject,
|
||||
onProgress: options.onProgress,
|
||||
@@ -137,6 +143,14 @@ export class WebRuntimeRpcClient {
|
||||
return
|
||||
}
|
||||
if (envelope.type === 'result') {
|
||||
const workspaceChange = getWorkspaceChange(envelope, pending)
|
||||
if (workspaceChange) {
|
||||
try {
|
||||
this.options.onWorkspaceChanged?.(workspaceChange)
|
||||
} catch (error) {
|
||||
console.error('Web runtime workspace change callback failed', error)
|
||||
}
|
||||
}
|
||||
this.settleRequest(envelope.id, 'resolve', envelope.payload.result)
|
||||
return
|
||||
}
|
||||
@@ -181,6 +195,29 @@ export class WebRuntimeRpcClient {
|
||||
}
|
||||
}
|
||||
|
||||
function getWorkspaceChange(
|
||||
envelope: Extract<RpcResponseEnvelope, { type: 'result' }>,
|
||||
pending: Pick<PendingRequest, 'taskType' | 'payload'>
|
||||
): WebRuntimeWorkspaceChangeEvent | null {
|
||||
const result = envelope.payload.result
|
||||
switch (envelope.payload.taskType) {
|
||||
case 'import.start':
|
||||
return { type: 'import', sessionId: (result as WebRuntimeTaskResult<'import.start'>).sessionId }
|
||||
case 'session.delete': {
|
||||
const deleted = result as WebRuntimeTaskResult<'session.delete'>
|
||||
const payload = pending.payload as WebRuntimeTaskPayload<'session.delete'>
|
||||
return deleted.deleted ? { type: 'delete', sessionId: payload.sessionId } : null
|
||||
}
|
||||
case 'session.rename': {
|
||||
const renamed = result as WebRuntimeTaskResult<'session.rename'>
|
||||
const payload = pending.payload as WebRuntimeTaskPayload<'session.rename'>
|
||||
return renamed.renamed ? { type: 'rename', sessionId: payload.sessionId } : null
|
||||
}
|
||||
default:
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
function createClientError(code: string, message: string): Error & { code: string } {
|
||||
return Object.assign(new Error(message), { code })
|
||||
}
|
||||
|
||||
@@ -229,6 +229,11 @@ export interface RuntimeLogEvent {
|
||||
data?: Record<string, unknown>
|
||||
}
|
||||
|
||||
export interface WebRuntimeWorkspaceChangeEvent {
|
||||
type: 'import' | 'delete' | 'rename'
|
||||
sessionId: string
|
||||
}
|
||||
|
||||
export interface RpcProgressEnvelope {
|
||||
id: string
|
||||
type: 'progress'
|
||||
|
||||
@@ -79,4 +79,27 @@ describe('SqliteWasmDatabaseAdapter', () => {
|
||||
db.close()
|
||||
}
|
||||
})
|
||||
|
||||
it('finalizes prepared statements before closing the database', async () => {
|
||||
const sqlite3 = await sqlite3InitModule()
|
||||
const rawDb = new sqlite3.oo1.DB(':memory:', 'c')
|
||||
const originalPrepare = rawDb.prepare.bind(rawDb)
|
||||
let finalizedStatements = 0
|
||||
|
||||
rawDb.prepare = ((sql) => {
|
||||
const statement = originalPrepare(sql)
|
||||
const originalFinalize = statement.finalize.bind(statement)
|
||||
statement.finalize = () => {
|
||||
finalizedStatements += 1
|
||||
return originalFinalize()
|
||||
}
|
||||
return statement
|
||||
}) as typeof rawDb.prepare
|
||||
|
||||
const db = new SqliteWasmDatabaseAdapter(sqlite3, rawDb)
|
||||
db.prepare('SELECT 1 AS value').get()
|
||||
db.close()
|
||||
|
||||
assert.equal(finalizedStatements, 1)
|
||||
})
|
||||
})
|
||||
@@ -8,6 +8,7 @@ import type {
|
||||
|
||||
class SqliteWasmPreparedStatement implements CorePreparedStatement {
|
||||
readonly?: boolean
|
||||
private finalized = false
|
||||
|
||||
constructor(
|
||||
private readonly sqlite3: Sqlite3Static,
|
||||
@@ -44,6 +45,12 @@ class SqliteWasmPreparedStatement implements CorePreparedStatement {
|
||||
})
|
||||
}
|
||||
|
||||
finalize(): void {
|
||||
if (this.finalized) return
|
||||
this.finalized = true
|
||||
this.statement.finalize()
|
||||
}
|
||||
|
||||
private withBindings<T>(params: unknown[], operation: () => T): T {
|
||||
try {
|
||||
if (params.length > 0) this.statement.bind(params as BindingSpec)
|
||||
@@ -56,6 +63,7 @@ class SqliteWasmPreparedStatement implements CorePreparedStatement {
|
||||
|
||||
export class SqliteWasmDatabaseAdapter implements DatabaseAdapter {
|
||||
readonly?: boolean
|
||||
private readonly statements = new Set<SqliteWasmPreparedStatement>()
|
||||
|
||||
constructor(
|
||||
private readonly sqlite3: Sqlite3Static,
|
||||
@@ -69,7 +77,9 @@ export class SqliteWasmDatabaseAdapter implements DatabaseAdapter {
|
||||
}
|
||||
|
||||
prepare(sql: string): CorePreparedStatement {
|
||||
return new SqliteWasmPreparedStatement(this.sqlite3, this.db, this.db.prepare(sql))
|
||||
const statement = new SqliteWasmPreparedStatement(this.sqlite3, this.db, this.db.prepare(sql))
|
||||
this.statements.add(statement)
|
||||
return statement
|
||||
}
|
||||
|
||||
transaction<T>(fn: () => T): T {
|
||||
@@ -93,6 +103,8 @@ export class SqliteWasmDatabaseAdapter implements DatabaseAdapter {
|
||||
}
|
||||
|
||||
close(): void {
|
||||
for (const statement of this.statements) statement.finalize()
|
||||
this.statements.clear()
|
||||
this.db.close()
|
||||
}
|
||||
}
|
||||
@@ -3,6 +3,7 @@ import { describe, it } from 'node:test'
|
||||
import sqlite3InitModule, { type Database, type SAHPoolUtil } from '@sqlite.org/sqlite-wasm'
|
||||
import { CURRENT_SCHEMA_VERSION, getHourlyActivity } from '@openchatlab/core'
|
||||
import { BrowserDatabaseRuntime } from './database-runtime'
|
||||
import type { WebRuntimeLockManager } from './workspace-lease'
|
||||
|
||||
async function createMemoryRuntime(options: { initializationFailures?: number } = {}): Promise<{
|
||||
runtime: BrowserDatabaseRuntime
|
||||
@@ -27,6 +28,9 @@ async function createMemoryRuntime(options: { initializationFailures?: number }
|
||||
unlink: (filename: string) => filenames.delete(filename),
|
||||
reserveMinimumCapacity: async (minimum: number) => minimum,
|
||||
getFileNames: () => [...filenames],
|
||||
pauseVfs: () => pool,
|
||||
unpauseVfs: async () => pool,
|
||||
isPaused: () => false,
|
||||
} as unknown as SAHPoolUtil
|
||||
let remainingInitializationFailures = options.initializationFailures ?? 0
|
||||
return {
|
||||
@@ -104,6 +108,8 @@ describe('BrowserDatabaseRuntime', () => {
|
||||
|
||||
assert.equal(count, 1)
|
||||
assert.deepEqual(stages, [
|
||||
'opfs-workspace-lock-waiting',
|
||||
'opfs-workspace-lock-acquired',
|
||||
'sqlite-initializing',
|
||||
'sqlite-ready',
|
||||
'opfs-pool-initializing',
|
||||
@@ -112,6 +118,9 @@ describe('BrowserDatabaseRuntime', () => {
|
||||
'opfs-database-opened',
|
||||
'schema-initializing',
|
||||
'schema-ready',
|
||||
'opfs-pool-pausing',
|
||||
'opfs-pool-paused',
|
||||
'opfs-workspace-lock-released',
|
||||
])
|
||||
assert.equal(openedDatabases[0].isOpen(), false)
|
||||
assert.deepEqual(await runtime.getDatabaseFilenames(), ['/catalog.db'])
|
||||
@@ -127,4 +136,133 @@ describe('BrowserDatabaseRuntime', () => {
|
||||
await assert.rejects(runtime.deleteDatabase('/held.db'), /must be closed/)
|
||||
await runtime.close()
|
||||
})
|
||||
|
||||
it('pauses the SAH pool after a scoped workspace task completes', async () => {
|
||||
const sqlite3 = await sqlite3InitModule()
|
||||
const events: string[] = []
|
||||
let paused = false
|
||||
class MemoryPoolDatabase {
|
||||
constructor() {
|
||||
return new sqlite3.oo1.DB(':memory:', 'c')
|
||||
}
|
||||
}
|
||||
const pool = {
|
||||
OpfsSAHPoolDb: MemoryPoolDatabase,
|
||||
pauseVfs() {
|
||||
events.push('paused')
|
||||
paused = true
|
||||
return pool
|
||||
},
|
||||
async unpauseVfs() {
|
||||
events.push('resumed')
|
||||
paused = false
|
||||
return pool
|
||||
},
|
||||
isPaused: () => paused,
|
||||
} as unknown as SAHPoolUtil
|
||||
const runtime = new BrowserDatabaseRuntime(async () => ({ sqlite3, pool }))
|
||||
|
||||
await runtime.withWorkspaceLease(async () => {
|
||||
events.push('first-task')
|
||||
})
|
||||
await runtime.withWorkspaceLease(async () => {
|
||||
events.push('second-task')
|
||||
})
|
||||
|
||||
assert.deepEqual(events, ['first-task', 'paused', 'resumed', 'second-task', 'paused'])
|
||||
})
|
||||
|
||||
it('serializes two runtimes so the first pool pauses before the second pool initializes', async () => {
|
||||
const sqlite3 = await sqlite3InitModule()
|
||||
const events: string[] = []
|
||||
const lockManager = new TestLockManager()
|
||||
let activePool: string | undefined
|
||||
|
||||
const createRuntime = (name: string) =>
|
||||
new BrowserDatabaseRuntime(async () => {
|
||||
if (activePool) throw new DOMException(`Pool ${activePool} is active`, 'NoModificationAllowedError')
|
||||
activePool = name
|
||||
events.push(`${name}-initialized`)
|
||||
let paused = false
|
||||
const pool = {
|
||||
isPaused: () => paused,
|
||||
pauseVfs() {
|
||||
paused = true
|
||||
activePool = undefined
|
||||
events.push(`${name}-paused`)
|
||||
return pool
|
||||
},
|
||||
async unpauseVfs() {
|
||||
if (activePool) throw new DOMException(`Pool ${activePool} is active`, 'NoModificationAllowedError')
|
||||
paused = false
|
||||
activePool = name
|
||||
events.push(`${name}-resumed`)
|
||||
return pool
|
||||
},
|
||||
} as unknown as SAHPoolUtil
|
||||
return { sqlite3, pool }
|
||||
}, lockManager)
|
||||
|
||||
const first = createRuntime('first')
|
||||
const second = createRuntime('second')
|
||||
let releaseFirst!: () => void
|
||||
const firstGate = new Promise<void>((resolve) => {
|
||||
releaseFirst = resolve
|
||||
})
|
||||
let firstStarted!: () => void
|
||||
const firstStartedPromise = new Promise<void>((resolve) => {
|
||||
firstStarted = resolve
|
||||
})
|
||||
|
||||
const firstTask = first.withWorkspaceLease(async () => {
|
||||
events.push('first-task')
|
||||
firstStarted()
|
||||
await firstGate
|
||||
})
|
||||
await firstStartedPromise
|
||||
|
||||
let secondRan = false
|
||||
const secondTask = second.withWorkspaceLease(async () => {
|
||||
secondRan = true
|
||||
events.push('second-task')
|
||||
})
|
||||
await new Promise<void>((resolve) => setImmediate(resolve))
|
||||
assert.equal(secondRan, false)
|
||||
|
||||
releaseFirst()
|
||||
await Promise.all([firstTask, secondTask])
|
||||
|
||||
assert.deepEqual(events, [
|
||||
'first-initialized',
|
||||
'first-task',
|
||||
'first-paused',
|
||||
'second-initialized',
|
||||
'second-task',
|
||||
'second-paused',
|
||||
])
|
||||
})
|
||||
})
|
||||
|
||||
class TestLockManager implements WebRuntimeLockManager {
|
||||
private active = false
|
||||
private readonly queue: Array<() => void> = []
|
||||
|
||||
request(_name: string, _options: { mode: 'exclusive' }, callback: (lock: object) => Promise<void>): Promise<void> {
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
const run = () => {
|
||||
this.active = true
|
||||
void callback({})
|
||||
.then(resolve, reject)
|
||||
.finally(() => {
|
||||
this.active = false
|
||||
this.queue.shift()?.()
|
||||
})
|
||||
}
|
||||
if (this.active) {
|
||||
this.queue.push(run)
|
||||
} else {
|
||||
run()
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -4,6 +4,12 @@ import { WebRuntimeError } from '../runtime-error'
|
||||
import type { WorkspaceDatabaseStage } from '../storage/workspace-database'
|
||||
import { SqliteWasmDatabaseAdapter } from './adapter'
|
||||
import { initializeOpfsSqlite, type InitializedSqliteRuntime, type SqliteInitializationStage } from './opfs'
|
||||
import {
|
||||
acquireWebRuntimeWorkspaceLease,
|
||||
getWebRuntimeLockManager,
|
||||
type WebRuntimeLockManager,
|
||||
type WebRuntimeWorkspaceLease,
|
||||
} from './workspace-lease'
|
||||
|
||||
export type DatabaseOpenStage = WorkspaceDatabaseStage
|
||||
|
||||
@@ -11,59 +17,94 @@ export class BrowserDatabaseRuntime {
|
||||
private initialized: Promise<InitializedSqliteRuntime> | undefined
|
||||
private database: DatabaseAdapter | undefined
|
||||
private filename: string | undefined
|
||||
private workspaceLease: WebRuntimeWorkspaceLease | undefined
|
||||
private workspaceRuntime: InitializedSqliteRuntime | undefined
|
||||
private workspaceLeaseDepth = 0
|
||||
|
||||
constructor(
|
||||
private readonly initializeSqlite: (
|
||||
onStage?: (stage: SqliteInitializationStage) => void
|
||||
) => Promise<InitializedSqliteRuntime> = initializeOpfsSqlite
|
||||
) => Promise<InitializedSqliteRuntime> = initializeOpfsSqlite,
|
||||
private readonly lockManager: WebRuntimeLockManager | undefined = getWebRuntimeLockManager()
|
||||
) {}
|
||||
|
||||
async open(filename: string, onStage?: (stage: DatabaseOpenStage) => void): Promise<OpenDatabaseResult> {
|
||||
validateDatabaseFilename(filename)
|
||||
return this.withWorkspaceLease(async () => {
|
||||
validateDatabaseFilename(filename)
|
||||
|
||||
if (this.database) {
|
||||
if (this.filename === filename) return this.buildOpenResult(filename, await this.getInitializedRuntime())
|
||||
throw new WebRuntimeError(
|
||||
'DATABASE_ALREADY_OPEN',
|
||||
`Database ${this.filename ?? '(unknown)'} is already open; close it before opening another database`
|
||||
)
|
||||
}
|
||||
|
||||
const runtime = await this.getInitializedRuntime(onStage)
|
||||
onStage?.('opfs-database-opening')
|
||||
const rawDatabase = new runtime.pool.OpfsSAHPoolDb(filename)
|
||||
const database = new SqliteWasmDatabaseAdapter(runtime.sqlite3, rawDatabase)
|
||||
onStage?.('opfs-database-opened')
|
||||
|
||||
try {
|
||||
database.pragma('foreign_keys = ON')
|
||||
onStage?.('schema-initializing')
|
||||
database.exec(CHAT_DB_TABLES)
|
||||
database.exec(CHAT_DB_INDEXES)
|
||||
onStage?.('schema-ready')
|
||||
} catch (error) {
|
||||
try {
|
||||
database.close()
|
||||
} catch {
|
||||
// Keep the schema error as the primary failure.
|
||||
if (this.database) {
|
||||
if (this.filename === filename) return this.buildOpenResult(filename, await this.getInitializedRuntime())
|
||||
throw new WebRuntimeError(
|
||||
'DATABASE_ALREADY_OPEN',
|
||||
`Database ${this.filename ?? '(unknown)'} is already open; close it before opening another database`
|
||||
)
|
||||
}
|
||||
throw new WebRuntimeError('SCHEMA_INITIALIZATION_FAILED', 'Could not initialize the browser database schema', {
|
||||
cause: error,
|
||||
})
|
||||
}
|
||||
|
||||
this.database = database
|
||||
this.filename = filename
|
||||
return this.buildOpenResult(filename, runtime)
|
||||
const runtime = await this.getInitializedRuntime(onStage)
|
||||
onStage?.('opfs-database-opening')
|
||||
const rawDatabase = new runtime.pool.OpfsSAHPoolDb(filename)
|
||||
const database = new SqliteWasmDatabaseAdapter(runtime.sqlite3, rawDatabase)
|
||||
onStage?.('opfs-database-opened')
|
||||
|
||||
try {
|
||||
database.pragma('foreign_keys = ON')
|
||||
onStage?.('schema-initializing')
|
||||
database.exec(CHAT_DB_TABLES)
|
||||
database.exec(CHAT_DB_INDEXES)
|
||||
onStage?.('schema-ready')
|
||||
} catch (error) {
|
||||
try {
|
||||
database.close()
|
||||
} catch {
|
||||
// Keep the schema error as the primary failure.
|
||||
}
|
||||
throw new WebRuntimeError('SCHEMA_INITIALIZATION_FAILED', 'Could not initialize the browser database schema', {
|
||||
cause: error,
|
||||
})
|
||||
}
|
||||
|
||||
this.database = database
|
||||
this.filename = filename
|
||||
return this.buildOpenResult(filename, runtime)
|
||||
}, onStage)
|
||||
}
|
||||
|
||||
async close(): Promise<{ closed: boolean }> {
|
||||
if (!this.database) return { closed: false }
|
||||
const database = this.database
|
||||
database.close()
|
||||
this.database = undefined
|
||||
this.filename = undefined
|
||||
return { closed: true }
|
||||
async close(onStage?: (stage: DatabaseOpenStage) => void): Promise<{ closed: boolean }> {
|
||||
return this.withWorkspaceLease(async () => {
|
||||
if (!this.database) return { closed: false }
|
||||
const database = this.database
|
||||
database.close()
|
||||
this.database = undefined
|
||||
this.filename = undefined
|
||||
return { closed: true }
|
||||
}, onStage)
|
||||
}
|
||||
|
||||
async withWorkspaceLease<T>(operation: () => Promise<T>, onStage?: (stage: DatabaseOpenStage) => void): Promise<T> {
|
||||
await this.acquireWorkspaceLease(onStage)
|
||||
this.workspaceLeaseDepth += 1
|
||||
let operationFailed = false
|
||||
let operationError: unknown
|
||||
let result!: T
|
||||
|
||||
try {
|
||||
result = await operation()
|
||||
} catch (error) {
|
||||
operationFailed = true
|
||||
operationError = error
|
||||
}
|
||||
|
||||
this.workspaceLeaseDepth -= 1
|
||||
if (this.workspaceLeaseDepth === 0 && !this.database) {
|
||||
try {
|
||||
await this.releaseWorkspaceLease(onStage)
|
||||
} catch (releaseError) {
|
||||
if (!operationFailed) throw releaseError
|
||||
}
|
||||
}
|
||||
|
||||
if (operationFailed) throw operationError
|
||||
return result
|
||||
}
|
||||
|
||||
async withDatabase<T>(
|
||||
@@ -72,60 +113,71 @@ export class BrowserDatabaseRuntime {
|
||||
operation: (db: DatabaseAdapter) => T,
|
||||
onStage?: (stage: WorkspaceDatabaseStage) => void
|
||||
): Promise<T> {
|
||||
validateDatabaseFilename(filename)
|
||||
if (this.database) {
|
||||
throw new WebRuntimeError(
|
||||
'DATABASE_ALREADY_OPEN',
|
||||
`Database ${this.filename ?? '(unknown)'} is already open; close it before running a workspace operation`
|
||||
)
|
||||
}
|
||||
return this.withWorkspaceLease(async () => {
|
||||
validateDatabaseFilename(filename)
|
||||
if (this.database) {
|
||||
throw new WebRuntimeError(
|
||||
'DATABASE_ALREADY_OPEN',
|
||||
`Database ${this.filename ?? '(unknown)'} is already open; close it before running a workspace operation`
|
||||
)
|
||||
}
|
||||
|
||||
const runtime = await this.getInitializedRuntime(onStage)
|
||||
onStage?.('opfs-database-opening')
|
||||
const rawDatabase = new runtime.pool.OpfsSAHPoolDb(filename)
|
||||
const database = new SqliteWasmDatabaseAdapter(runtime.sqlite3, rawDatabase)
|
||||
onStage?.('opfs-database-opened')
|
||||
let operationFailed = false
|
||||
let operationError: unknown
|
||||
let result!: T
|
||||
const runtime = await this.getInitializedRuntime(onStage)
|
||||
onStage?.('opfs-database-opening')
|
||||
const rawDatabase = new runtime.pool.OpfsSAHPoolDb(filename)
|
||||
const database = new SqliteWasmDatabaseAdapter(runtime.sqlite3, rawDatabase)
|
||||
onStage?.('opfs-database-opened')
|
||||
let operationFailed = false
|
||||
let operationError: unknown
|
||||
let result!: T
|
||||
|
||||
try {
|
||||
database.pragma('foreign_keys = ON')
|
||||
onStage?.('schema-initializing')
|
||||
database.exec(schemaSql)
|
||||
onStage?.('schema-ready')
|
||||
result = operation(database)
|
||||
} catch (error) {
|
||||
operationFailed = true
|
||||
operationError = error
|
||||
}
|
||||
try {
|
||||
database.pragma('foreign_keys = ON')
|
||||
onStage?.('schema-initializing')
|
||||
database.exec(schemaSql)
|
||||
onStage?.('schema-ready')
|
||||
result = operation(database)
|
||||
} catch (error) {
|
||||
operationFailed = true
|
||||
operationError = error
|
||||
}
|
||||
|
||||
try {
|
||||
database.close()
|
||||
} catch (closeError) {
|
||||
if (!operationFailed) throw closeError
|
||||
}
|
||||
if (operationFailed) throw operationError
|
||||
return result
|
||||
try {
|
||||
database.close()
|
||||
} catch (closeError) {
|
||||
if (!operationFailed) throw closeError
|
||||
}
|
||||
if (operationFailed) throw operationError
|
||||
return result
|
||||
}, onStage)
|
||||
}
|
||||
|
||||
async deleteDatabase(filename: string): Promise<boolean> {
|
||||
validateDatabaseFilename(filename)
|
||||
if (this.filename === filename) {
|
||||
throw new WebRuntimeError('DATABASE_ALREADY_OPEN', `Database ${filename} must be closed before it can be deleted`)
|
||||
}
|
||||
const runtime = await this.getInitializedRuntime()
|
||||
return runtime.pool.unlink(filename)
|
||||
return this.withWorkspaceLease(async () => {
|
||||
validateDatabaseFilename(filename)
|
||||
if (this.filename === filename) {
|
||||
throw new WebRuntimeError(
|
||||
'DATABASE_ALREADY_OPEN',
|
||||
`Database ${filename} must be closed before it can be deleted`
|
||||
)
|
||||
}
|
||||
const runtime = await this.getInitializedRuntime()
|
||||
return runtime.pool.unlink(filename)
|
||||
})
|
||||
}
|
||||
|
||||
async ensureCapacity(minimum: number): Promise<number> {
|
||||
const runtime = await this.getInitializedRuntime()
|
||||
return runtime.pool.reserveMinimumCapacity(minimum)
|
||||
return this.withWorkspaceLease(async () => {
|
||||
const runtime = await this.getInitializedRuntime()
|
||||
return runtime.pool.reserveMinimumCapacity(minimum)
|
||||
})
|
||||
}
|
||||
|
||||
async getDatabaseFilenames(): Promise<string[]> {
|
||||
const runtime = await this.getInitializedRuntime()
|
||||
return runtime.pool.getFileNames()
|
||||
return this.withWorkspaceLease(async () => {
|
||||
const runtime = await this.getInitializedRuntime()
|
||||
return runtime.pool.getFileNames()
|
||||
})
|
||||
}
|
||||
|
||||
getOpenDatabase(): DatabaseAdapter {
|
||||
@@ -145,6 +197,45 @@ export class BrowserDatabaseRuntime {
|
||||
}
|
||||
}
|
||||
|
||||
private async acquireWorkspaceLease(onStage?: (stage: DatabaseOpenStage) => void): Promise<void> {
|
||||
if (this.workspaceLease) return
|
||||
|
||||
onStage?.('opfs-workspace-lock-waiting')
|
||||
const lease = await acquireWebRuntimeWorkspaceLease(this.lockManager)
|
||||
|
||||
try {
|
||||
onStage?.('opfs-workspace-lock-acquired')
|
||||
const runtime = await this.getInitializedRuntime(onStage)
|
||||
if (runtime.pool.isPaused()) {
|
||||
onStage?.('opfs-pool-resuming')
|
||||
await runtime.pool.unpauseVfs()
|
||||
onStage?.('opfs-pool-resumed')
|
||||
}
|
||||
this.workspaceLease = lease
|
||||
this.workspaceRuntime = runtime
|
||||
} catch (error) {
|
||||
lease.release()
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
private async releaseWorkspaceLease(onStage?: (stage: DatabaseOpenStage) => void): Promise<void> {
|
||||
const lease = this.workspaceLease
|
||||
const runtime = this.workspaceRuntime
|
||||
if (!lease) return
|
||||
|
||||
if (runtime && !runtime.pool.isPaused()) {
|
||||
onStage?.('opfs-pool-pausing')
|
||||
runtime.pool.pauseVfs()
|
||||
onStage?.('opfs-pool-paused')
|
||||
}
|
||||
|
||||
this.workspaceLease = undefined
|
||||
this.workspaceRuntime = undefined
|
||||
lease.release()
|
||||
onStage?.('opfs-workspace-lock-released')
|
||||
}
|
||||
|
||||
private buildOpenResult(filename: string, runtime: InitializedSqliteRuntime): OpenDatabaseResult {
|
||||
return {
|
||||
filename,
|
||||
|
||||
@@ -0,0 +1,60 @@
|
||||
import assert from 'node:assert/strict'
|
||||
import { describe, it } from 'node:test'
|
||||
import {
|
||||
WEB_RUNTIME_WORKSPACE_LOCK,
|
||||
acquireWebRuntimeWorkspaceLease,
|
||||
type WebRuntimeLockManager,
|
||||
} from './workspace-lease'
|
||||
|
||||
class QueuedLockManager implements WebRuntimeLockManager {
|
||||
private active = false
|
||||
private readonly queue: Array<() => void> = []
|
||||
|
||||
request(name: string, options: { mode: 'exclusive' }, callback: (lock: object) => Promise<void>): Promise<void> {
|
||||
assert.equal(name, WEB_RUNTIME_WORKSPACE_LOCK)
|
||||
assert.deepEqual(options, { mode: 'exclusive' })
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
const run = () => {
|
||||
this.active = true
|
||||
void callback({})
|
||||
.then(resolve, reject)
|
||||
.finally(() => {
|
||||
this.active = false
|
||||
this.queue.shift()?.()
|
||||
})
|
||||
}
|
||||
if (this.active) {
|
||||
this.queue.push(run)
|
||||
} else {
|
||||
run()
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
describe('acquireWebRuntimeWorkspaceLease', () => {
|
||||
it('queues a second workspace lease until the first lease is released', async () => {
|
||||
const manager = new QueuedLockManager()
|
||||
const first = await acquireWebRuntimeWorkspaceLease(manager)
|
||||
let secondAcquired = false
|
||||
const secondPromise = acquireWebRuntimeWorkspaceLease(manager).then((lease) => {
|
||||
secondAcquired = true
|
||||
return lease
|
||||
})
|
||||
|
||||
await new Promise<void>((resolve) => setImmediate(resolve))
|
||||
assert.equal(secondAcquired, false)
|
||||
|
||||
first.release()
|
||||
const second = await secondPromise
|
||||
assert.equal(secondAcquired, true)
|
||||
second.release()
|
||||
})
|
||||
|
||||
it('returns a no-op lease when Web Locks is unavailable', async () => {
|
||||
const lease = await acquireWebRuntimeWorkspaceLease(undefined)
|
||||
|
||||
assert.doesNotThrow(() => lease.release())
|
||||
assert.doesNotThrow(() => lease.release())
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,49 @@
|
||||
export const WEB_RUNTIME_WORKSPACE_LOCK = 'chatlab-web-wasm-opfs-workspace'
|
||||
|
||||
export interface WebRuntimeLockManager {
|
||||
request(name: string, options: { mode: 'exclusive' }, callback: (lock: object) => Promise<void>): Promise<void>
|
||||
}
|
||||
|
||||
export interface WebRuntimeWorkspaceLease {
|
||||
release(): void
|
||||
}
|
||||
|
||||
export function getWebRuntimeLockManager(): WebRuntimeLockManager | undefined {
|
||||
if (typeof navigator === 'undefined') return undefined
|
||||
return (navigator as Navigator & { locks?: WebRuntimeLockManager }).locks
|
||||
}
|
||||
|
||||
export async function acquireWebRuntimeWorkspaceLease(
|
||||
lockManager: WebRuntimeLockManager | undefined = getWebRuntimeLockManager()
|
||||
): Promise<WebRuntimeWorkspaceLease> {
|
||||
if (!lockManager) return { release: () => undefined }
|
||||
|
||||
let releaseLock: () => void = () => undefined
|
||||
const released = new Promise<void>((resolve) => {
|
||||
releaseLock = resolve
|
||||
})
|
||||
let releaseCalled = false
|
||||
const lease: WebRuntimeWorkspaceLease = {
|
||||
release() {
|
||||
if (releaseCalled) return
|
||||
releaseCalled = true
|
||||
releaseLock()
|
||||
},
|
||||
}
|
||||
|
||||
let resolveAcquired!: (lease: WebRuntimeWorkspaceLease) => void
|
||||
let rejectAcquired!: (error: unknown) => void
|
||||
const acquired = new Promise<WebRuntimeWorkspaceLease>((resolve, reject) => {
|
||||
resolveAcquired = resolve
|
||||
rejectAcquired = reject
|
||||
})
|
||||
|
||||
void lockManager
|
||||
.request(WEB_RUNTIME_WORKSPACE_LOCK, { mode: 'exclusive' }, async () => {
|
||||
resolveAcquired(lease)
|
||||
await released
|
||||
})
|
||||
.catch(rejectAcquired)
|
||||
|
||||
return acquired
|
||||
}
|
||||
@@ -1,14 +1,21 @@
|
||||
import type { DatabaseAdapter } from '@openchatlab/core'
|
||||
|
||||
export type WorkspaceDatabaseStage =
|
||||
| 'opfs-workspace-lock-waiting'
|
||||
| 'opfs-workspace-lock-acquired'
|
||||
| 'sqlite-initializing'
|
||||
| 'sqlite-ready'
|
||||
| 'opfs-pool-initializing'
|
||||
| 'opfs-pool-ready'
|
||||
| 'opfs-pool-resuming'
|
||||
| 'opfs-pool-resumed'
|
||||
| 'opfs-database-opening'
|
||||
| 'opfs-database-opened'
|
||||
| 'schema-initializing'
|
||||
| 'schema-ready'
|
||||
| 'opfs-pool-pausing'
|
||||
| 'opfs-pool-paused'
|
||||
| 'opfs-workspace-lock-released'
|
||||
|
||||
export interface WorkspaceDatabasePort {
|
||||
withDatabase<T>(
|
||||
|
||||
@@ -37,6 +37,12 @@ class FakeDatabaseRuntime implements WorkerDatabaseRuntime {
|
||||
openCalls: string[] = []
|
||||
closeCalls = 0
|
||||
closeGate: Promise<void> | undefined
|
||||
workspaceLeaseCalls = 0
|
||||
|
||||
async withWorkspaceLease<T>(operation: () => Promise<T>): Promise<T> {
|
||||
this.workspaceLeaseCalls += 1
|
||||
return operation()
|
||||
}
|
||||
|
||||
async open(filename: string, onStage?: (stage: DatabaseOpenStage) => void) {
|
||||
this.openCalls.push(filename)
|
||||
@@ -594,6 +600,7 @@ describe('WebRuntimeWorkerController', () => {
|
||||
.map((message) => (message.type === 'log' ? message.payload.message : '')),
|
||||
['sqlite-ready', 'opfs-pool-ready', 'opfs-database-opened', 'schema-ready']
|
||||
)
|
||||
assert.equal(database.workspaceLeaseCalls, 14)
|
||||
})
|
||||
|
||||
it('routes the complete Insights query set through the Worker runtime', async () => {
|
||||
|
||||
@@ -20,8 +20,9 @@ export interface WorkerMessageSink {
|
||||
}
|
||||
|
||||
export interface WorkerDatabaseRuntime extends WorkspaceDatabasePort {
|
||||
withWorkspaceLease<T>(operation: () => Promise<T>, onStage?: (stage: DatabaseOpenStage) => void): Promise<T>
|
||||
open(filename: string, onStage?: (stage: DatabaseOpenStage) => void): Promise<OpenDatabaseResult>
|
||||
close(): Promise<{ closed: boolean }>
|
||||
close(onStage?: (stage: DatabaseOpenStage) => void): Promise<{ closed: boolean }>
|
||||
}
|
||||
|
||||
export type WorkerSessionRuntime = Pick<
|
||||
@@ -84,7 +85,12 @@ export class WebRuntimeWorkerController {
|
||||
private async execute(request: RpcRequestEnvelope): Promise<void> {
|
||||
try {
|
||||
this.throwIfCancelled(request.id)
|
||||
const result = await this.dispatch(request)
|
||||
const result = requiresWorkspaceLease(request.type)
|
||||
? await this.databaseRuntime.withWorkspaceLease(
|
||||
() => this.dispatch(request),
|
||||
(stage) => this.handleDatabaseStage(request, stage)
|
||||
)
|
||||
: await this.dispatch(request)
|
||||
this.sink.postMessage({
|
||||
id: request.id,
|
||||
type: 'result',
|
||||
@@ -134,7 +140,7 @@ export class WebRuntimeWorkerController {
|
||||
return result
|
||||
}
|
||||
case 'db.close': {
|
||||
const result = await this.databaseRuntime.close()
|
||||
const result = await this.databaseRuntime.close((stage) => this.handleDatabaseStage(request, stage))
|
||||
if (result.closed) {
|
||||
this.emitLog(request.id, {
|
||||
level: 'info',
|
||||
@@ -471,17 +477,26 @@ export class WebRuntimeWorkerController {
|
||||
}
|
||||
|
||||
private handleDatabaseStage(request: RpcRequestEnvelope, stage: DatabaseOpenStage): void {
|
||||
this.throwIfCancelled(request.id)
|
||||
const progressByStage: Record<DatabaseOpenStage, number> = {
|
||||
'opfs-workspace-lock-waiting': 0.05,
|
||||
'opfs-workspace-lock-acquired': 0.1,
|
||||
'sqlite-initializing': 0.2,
|
||||
'sqlite-ready': 0.35,
|
||||
'opfs-pool-initializing': 0.45,
|
||||
'opfs-pool-ready': 0.6,
|
||||
'opfs-pool-resuming': 0.65,
|
||||
'opfs-pool-resumed': 0.68,
|
||||
'opfs-database-opening': 0.7,
|
||||
'opfs-database-opened': 0.8,
|
||||
'schema-initializing': 0.9,
|
||||
'schema-ready': 1,
|
||||
'opfs-pool-pausing': 1,
|
||||
'opfs-pool-paused': 1,
|
||||
'opfs-workspace-lock-released': 1,
|
||||
}
|
||||
const isReleaseStage =
|
||||
stage === 'opfs-pool-pausing' || stage === 'opfs-pool-paused' || stage === 'opfs-workspace-lock-released'
|
||||
if (!isReleaseStage) this.throwIfCancelled(request.id)
|
||||
this.emitProgress(request, stage, progressByStage[stage])
|
||||
if (
|
||||
stage === 'sqlite-ready' ||
|
||||
@@ -495,6 +510,13 @@ export class WebRuntimeWorkerController {
|
||||
message: stage,
|
||||
})
|
||||
}
|
||||
if (stage === 'opfs-workspace-lock-acquired' || stage === 'opfs-pool-resumed' || stage === 'opfs-pool-paused') {
|
||||
this.emitLog(request.id, {
|
||||
level: 'debug',
|
||||
scope: 'web-runtime',
|
||||
message: stage,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
private emitProgress(request: RpcRequestEnvelope, stage: string, progress: number, messagesProcessed?: number): void {
|
||||
@@ -524,6 +546,15 @@ export class WebRuntimeWorkerController {
|
||||
}
|
||||
}
|
||||
|
||||
function requiresWorkspaceLease(type: WebRuntimeTaskType): boolean {
|
||||
return (
|
||||
type !== 'capabilities.check' &&
|
||||
type !== 'import.formats' &&
|
||||
type !== 'import.detectFormat' &&
|
||||
type !== 'import.scanChats'
|
||||
)
|
||||
}
|
||||
|
||||
function serializeRpcError(error: unknown): SerializedRpcError {
|
||||
if (error instanceof WebRuntimeError) {
|
||||
return {
|
||||
|
||||
@@ -11,7 +11,7 @@
|
||||
"done": "Done",
|
||||
"retry": "Retry",
|
||||
"initFailed": "App initialization failed",
|
||||
"webWasmWorkspaceBusy": "Web WASM is already open in another tab. Close the other ChatLab Web WASM tab and retry.",
|
||||
"webWasmWorkspaceBusy": "The local workspace is temporarily in use by another tab. Wait a moment and retry.",
|
||||
"loading": "Loading...",
|
||||
"initializing": "Initializing...",
|
||||
"noResults": "No results found",
|
||||
|
||||
@@ -11,7 +11,7 @@
|
||||
"done": "完了",
|
||||
"retry": "リトライ",
|
||||
"initFailed": "アプリの初期化に失敗しました",
|
||||
"webWasmWorkspaceBusy": "Web WASM は別のタブで開かれています。もう一方の ChatLab Web WASM タブを閉じてから再試行してください。",
|
||||
"webWasmWorkspaceBusy": "ローカルワークスペースは一時的に別のタブで使用されています。しばらく待ってから再試行してください。",
|
||||
"loading": "読み込み中...",
|
||||
"initializing": "初期化中...",
|
||||
"noResults": "結果が見つかりません",
|
||||
|
||||
@@ -11,7 +11,7 @@
|
||||
"done": "完成",
|
||||
"retry": "重试",
|
||||
"initFailed": "应用初始化失败",
|
||||
"webWasmWorkspaceBusy": "Web WASM 已在另一个标签页中打开。请关闭另一个 ChatLab Web WASM 标签页后重试。",
|
||||
"webWasmWorkspaceBusy": "本地工作区暂时正由另一个标签页使用。请稍候后重试。",
|
||||
"loading": "加载中...",
|
||||
"initializing": "初始化中...",
|
||||
"noResults": "无匹配结果",
|
||||
|
||||
@@ -11,7 +11,7 @@
|
||||
"done": "完成",
|
||||
"retry": "重試",
|
||||
"initFailed": "應用程式初始化失敗",
|
||||
"webWasmWorkspaceBusy": "Web WASM 已在另一個分頁中開啟。請關閉另一個 ChatLab Web WASM 分頁後重試。",
|
||||
"webWasmWorkspaceBusy": "本機工作區暫時正由另一個分頁使用。請稍候後重試。",
|
||||
"loading": "載入中...",
|
||||
"initializing": "初始化中...",
|
||||
"noResults": "無匹配結果",
|
||||
|
||||
@@ -11,6 +11,7 @@ describe('registerWebWasmAdapters', () => {
|
||||
it('registers the browser runtime adapter and forwards Worker logs', () => {
|
||||
const registrations = new Map<string, unknown>()
|
||||
const logs: RuntimeLogEvent[] = []
|
||||
const workspaceChanges: string[] = []
|
||||
let clientOptions: WebRuntimeRpcClientOptions | undefined
|
||||
const client = {
|
||||
request: () => Promise.reject(new Error('not used')),
|
||||
@@ -24,6 +25,7 @@ describe('registerWebWasmAdapters', () => {
|
||||
return client
|
||||
},
|
||||
reportLog: (event) => logs.push(event),
|
||||
onWorkspaceChanged: (event) => workspaceChanges.push(`${event.type}:${event.sessionId}`),
|
||||
})
|
||||
|
||||
assert.deepEqual([...registrations.keys()], ['browser-runtime', 'import', 'data', 'platform', 'preferences'])
|
||||
@@ -38,6 +40,8 @@ describe('registerWebWasmAdapters', () => {
|
||||
message: 'Worker initialized',
|
||||
}
|
||||
clientOptions?.onLog?.(event)
|
||||
clientOptions?.onWorkspaceChanged?.({ type: 'rename', sessionId: 'session-one' })
|
||||
assert.deepEqual(logs, [event])
|
||||
assert.deepEqual(workspaceChanges, ['rename:session-one'])
|
||||
})
|
||||
})
|
||||
@@ -1,4 +1,9 @@
|
||||
import { createWebRuntimeClient, type RuntimeLogEvent, type WebRuntimeRpcClientOptions } from '@openchatlab/web-runtime'
|
||||
import {
|
||||
createWebRuntimeClient,
|
||||
type RuntimeLogEvent,
|
||||
type WebRuntimeRpcClientOptions,
|
||||
type WebRuntimeWorkspaceChangeEvent,
|
||||
} from '@openchatlab/web-runtime'
|
||||
import { reportRuntimeLog } from '@/services/log-report'
|
||||
import { BrowserRuntimeAdapter } from './browser'
|
||||
import type { BrowserRuntimeRpcPort } from './types'
|
||||
@@ -11,12 +16,13 @@ export interface RegisterWebWasmAdaptersOptions {
|
||||
register(key: string, adapter: unknown): void
|
||||
createClient?: (options: WebRuntimeRpcClientOptions) => BrowserRuntimeRpcPort
|
||||
reportLog?: (event: RuntimeLogEvent) => void
|
||||
onWorkspaceChanged?: (event: WebRuntimeWorkspaceChangeEvent) => void
|
||||
}
|
||||
|
||||
export function registerWebWasmAdapters(options: RegisterWebWasmAdaptersOptions): void {
|
||||
const reportLog = options.reportLog ?? reportRuntimeLog
|
||||
const createClient = options.createClient ?? createWebRuntimeClient
|
||||
const client = createClient({ onLog: reportLog })
|
||||
const client = createClient({ onLog: reportLog, onWorkspaceChanged: options.onWorkspaceChanged })
|
||||
|
||||
options.register('browser-runtime', new BrowserRuntimeAdapter(client))
|
||||
options.register('import', new BrowserImportAdapter(client))
|
||||
|
||||
Reference in new issue
Block a user