fix(tsconnect-worker): address round-3 review feedback
Bugs:
- Liveness detection now fires for all clients, including those that connect
while init is pending. Previously, the "hello" message was consumed by the
onFirst handler before registerClient set up its port.onmessage listener,
so navigator.locks.request was never called for the first tab and any tab
that connected during init. Fix: unified onHello handler in onconnect
extracts lockName from every "hello"; registerClient always takes lockName
and acquires the lock immediately.
- Remove dead bodyCallbacks field from bridgeDriveHandler pending map. The
map was allocated on every request but never read or populated; body chunk
reads go directly through entry.req.readBodyChunk().
- Drive cleanup in cleanupClient now only installs the no-op handler when no
other client still has driveRegistered = true, preventing the surviving
client's handler from being silently replaced.
Code quality:
- wrapDispatch: replace (s as any).dispatch with a two-step cast through
unknown — avoids the any escape hatch while satisfying RTK's overloaded
ThunkDispatch type.
- IndexedDBState.setState: add tx.onerror handler so failed IDB writes are
logged rather than silently dropped.
- pumpStreamToPort fire-and-forget in openWaitingFile fallback: add
.catch(() => {}) to make the intentional discard explicit.
- tsconfig.json: add skipLibCheck: true to match tsconfig.worker.json (both
transitively import @webnet/tsconnect which has an unresolved wasm_exec.js
declaration issue).
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -41,8 +41,8 @@ let store: ReturnType<typeof buildIpnStore> | null = null
|
||||
const clients = new Map<number, ClientEntry>()
|
||||
let nextClientId = 0
|
||||
|
||||
// Clients that connected before init completed.
|
||||
const pendingClients: Array<{ port: MessagePort; config?: WorkerConfig }> = []
|
||||
// Clients that connected before init completed, with their lock name already extracted.
|
||||
const pendingClients: Array<{ port: MessagePort; lockName: string }> = []
|
||||
let initState: "idle" | "pending" | "ready" | "failed" = "idle"
|
||||
let initError = ""
|
||||
|
||||
@@ -65,13 +65,13 @@ function errMsg(e: unknown): string {
|
||||
// ── Broadcast all store actions to connected clients ─────────────────────────
|
||||
|
||||
function wrapDispatch(s: ReturnType<typeof buildIpnStore>): void {
|
||||
const orig = s.dispatch.bind(s)
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
;(s as any).dispatch = (action: IpnAction) => {
|
||||
const orig = s.dispatch
|
||||
const wrapped = (action: Parameters<typeof orig>[0]) => {
|
||||
const result = orig(action)
|
||||
broadcast({ type: "action", action })
|
||||
return result
|
||||
broadcast({ type: "action", action: action as IpnAction })
|
||||
return result as ReturnType<typeof orig>
|
||||
}
|
||||
s.dispatch = wrapped as unknown as typeof orig
|
||||
}
|
||||
|
||||
// ── Worker-side Conn bridge ──────────────────────────────────────────────────
|
||||
@@ -205,7 +205,6 @@ function bridgeDriveHandler(): { port: MessagePort; handler: RawDriveHandler } {
|
||||
req: Parameters<RawDriveHandler>[0]
|
||||
res: Parameters<RawDriveHandler>[1]
|
||||
resolve: () => void
|
||||
bodyCallbacks: Map<number, (chunk: Uint8Array | null) => void>
|
||||
}
|
||||
>()
|
||||
|
||||
@@ -241,7 +240,7 @@ function bridgeDriveHandler(): { port: MessagePort; handler: RawDriveHandler } {
|
||||
const handler: RawDriveHandler = (req, res, permissions) => {
|
||||
const reqId = nextReqId++
|
||||
return new Promise<void>((resolve) => {
|
||||
pending.set(reqId, { req, res, resolve, bodyCallbacks: new Map() })
|
||||
pending.set(reqId, { req, res, resolve })
|
||||
const msg: DriveW2C = {
|
||||
type: "request",
|
||||
reqId,
|
||||
@@ -378,7 +377,7 @@ async function handleCall(
|
||||
// ReadableStream not transferable (Safari < 17): fall back to port pump
|
||||
const ch = new MessageChannel()
|
||||
send(port, { type: "streamReturn", id, port: ch.port2 } satisfies W2C, [ch.port2])
|
||||
pumpStreamToPort(stream, ch.port1)
|
||||
pumpStreamToPort(stream, ch.port1).catch(() => {})
|
||||
}
|
||||
break
|
||||
}
|
||||
@@ -551,10 +550,13 @@ function cleanupClient(clientId: number): void {
|
||||
}
|
||||
|
||||
if (entry.driveRegistered && ipn) {
|
||||
try {
|
||||
ipn.serveDrive(() => Promise.resolve())
|
||||
} catch {
|
||||
/* ipn may be shut down */
|
||||
const anotherHasDrive = [...clients.values()].some((c) => c.driveRegistered)
|
||||
if (!anotherHasDrive) {
|
||||
try {
|
||||
ipn.serveDrive(() => Promise.resolve())
|
||||
} catch {
|
||||
/* ipn may be shut down */
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -567,7 +569,7 @@ function cleanupClient(clientId: number): void {
|
||||
|
||||
// ── Client registration ───────────────────────────────────────────────────────
|
||||
|
||||
function registerClient(port: MessagePort): void {
|
||||
function registerClient(port: MessagePort, lockName: string): void {
|
||||
const clientId = nextClientId++
|
||||
const entry: ClientEntry = {
|
||||
port,
|
||||
@@ -583,20 +585,21 @@ function registerClient(port: MessagePort): void {
|
||||
send(port, { type: "preloadState", state: store.getState() })
|
||||
}
|
||||
|
||||
// Acquire the lock immediately — lockName is already known from the "hello" message,
|
||||
// which was consumed either here (ready path) or in onHello (pending path).
|
||||
navigator.locks.request(lockName, { mode: "exclusive" }, () => {
|
||||
cleanupClient(clientId)
|
||||
return Promise.resolve()
|
||||
})
|
||||
|
||||
send(port, { type: "ready" })
|
||||
|
||||
port.onmessage = (e: MessageEvent) => {
|
||||
const msg = e.data as C2W
|
||||
if (msg.type === "hello") {
|
||||
const { lockName } = msg
|
||||
navigator.locks.request(lockName, { mode: "exclusive" }, () => {
|
||||
cleanupClient(clientId)
|
||||
return Promise.resolve()
|
||||
})
|
||||
send(port, { type: "ready" })
|
||||
} else if (msg.type === "call") {
|
||||
if (msg.type === "call") {
|
||||
handleCall(port, entry, msg.id, msg.method, msg.args)
|
||||
}
|
||||
}
|
||||
port.start()
|
||||
}
|
||||
|
||||
// ── Initialization ────────────────────────────────────────────────────────────
|
||||
@@ -643,35 +646,36 @@ async function init(config: WorkerConfig): Promise<void> {
|
||||
sw.onconnect = (e: MessageEvent) => {
|
||||
const port = (e as MessageEvent & { ports: MessagePort[] }).ports[0]
|
||||
|
||||
if (initState === "ready") {
|
||||
registerClient(port)
|
||||
return
|
||||
}
|
||||
if (initState === "failed") {
|
||||
send(port, { type: "initError", error: initError })
|
||||
return
|
||||
}
|
||||
|
||||
// Capture the first message to determine config, then register.
|
||||
const onFirst = async (ev: MessageEvent) => {
|
||||
port.removeEventListener("message", onFirst)
|
||||
// Always wait for "hello" first so we have the lockName before registering.
|
||||
const onHello = async (ev: MessageEvent) => {
|
||||
port.removeEventListener("message", onHello)
|
||||
const msg = ev.data as C2W
|
||||
if (msg.type !== "hello") {
|
||||
send(port, { type: "initError", error: "expected hello as first message" })
|
||||
return
|
||||
}
|
||||
|
||||
if (initState === "ready") {
|
||||
registerClient(port, msg.lockName)
|
||||
return
|
||||
}
|
||||
|
||||
if (initState === "idle") {
|
||||
if (!msg.config) {
|
||||
send(port, { type: "initError", error: "init config required for first connection" })
|
||||
return
|
||||
}
|
||||
pendingClients.push({ port, config: msg.config })
|
||||
pendingClients.push({ port, lockName: msg.lockName })
|
||||
const localConfig = msg.config
|
||||
try {
|
||||
await init(localConfig)
|
||||
for (const pending of pendingClients) {
|
||||
registerClient(pending.port)
|
||||
registerClient(pending.port, pending.lockName)
|
||||
}
|
||||
} catch (err) {
|
||||
const error = errMsg(err)
|
||||
@@ -680,11 +684,12 @@ sw.onconnect = (e: MessageEvent) => {
|
||||
}
|
||||
}
|
||||
pendingClients.length = 0
|
||||
} else if (initState === "pending") {
|
||||
pendingClients.push({ port, config: msg.config })
|
||||
} else {
|
||||
// initState === "pending": another tab is already initialising; queue.
|
||||
pendingClients.push({ port, lockName: msg.lockName })
|
||||
}
|
||||
}
|
||||
|
||||
port.addEventListener("message", onFirst)
|
||||
port.addEventListener("message", onHello)
|
||||
port.start()
|
||||
}
|
||||
|
||||
@@ -8,7 +8,8 @@
|
||||
"outDir": "dist",
|
||||
"rootDir": "src",
|
||||
"sourceMap": true,
|
||||
"lib": ["ES2018", "DOM"]
|
||||
"lib": ["ES2018", "DOM"],
|
||||
"skipLibCheck": true
|
||||
},
|
||||
"include": ["src/**/*"],
|
||||
"exclude": ["src/worker.ts"]
|
||||
|
||||
@@ -178,6 +178,7 @@ export class IndexedDBState implements IPNStateStorage {
|
||||
this.#cache.set(id, value)
|
||||
const tx = this.#db.transaction([this.#storeName], "readwrite")
|
||||
tx.objectStore(this.#storeName).put(value, id)
|
||||
tx.onerror = () => console.error("IndexedDBState: write failed", tx.error)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user