Spaces:

wuhp
/

webgputest

Running

Deploy GitHub 6fdad9a

807bea4 verified about 7 hours ago

21.8 kB

const MAX_ARTIFACT_BYTES = 8 * 1024 * 1024
const CHUNK_BYTES = 16 * 1024
const MAX_FRAME_CHARS = 24 * 1024
const BUFFER_LIMIT_BYTES = 256 * 1024
const MAX_CONCURRENT_TRANSFERS = 2

type SignalPayload = { description?: RTCSessionDescriptionInit; candidate?: RTCIceCandidateInit }
type TransferStart = { type: 'artifact_start'; transfer_id: string; name: string; total_bytes: number; chunks: number; sha256: string }
type TransferChunk = { type: 'artifact_chunk'; transfer_id: string; index: number; data: string }
type IncomingTransfer = { peerId: string; start: TransferStart; chunks: Map<number, Uint8Array>; bytes: number; lastActivity: number }
type PendingTransfer = { peerId: string; resolve: (digest: string) => void; reject: (error: Error) => void; timer: number }
type PendingArtifactRead = { peerId: string; name: string; resolve: (bytes: Uint8Array) => void; reject: (error: Error) => void; timer: number }

function encode(bytes: Uint8Array) {
  let binary = ''
  for (let offset = 0; offset < bytes.length; offset += 0x8000) binary += String.fromCharCode(...bytes.subarray(offset, offset + 0x8000))
  return btoa(binary)
}

function decode(base64: string): Uint8Array {
  const binary = atob(base64)
  const result = new Uint8Array(binary.length)
  for (let i = 0; i < binary.length; i++) result[i] = binary.charCodeAt(i)
  return result
}

async function digest(bytes: Uint8Array) {
  const copy = bytes.slice().buffer
  const result = await crypto.subtle.digest('SHA-256', copy)
  return Array.from(new Uint8Array(result), byte => byte.toString(16).padStart(2, '0')).join('')
}

async function persistArtifact(key: string, value: { name: string; bytes: Uint8Array; sha256: string; receivedAt: number }) {
  const db = await new Promise<IDBDatabase>((resolve, reject) => {
    const request = indexedDB.open('mesh-peer-artifacts', 1)
    request.onupgradeneeded = () => request.result.createObjectStore('artifacts')
    request.onsuccess = () => resolve(request.result)
    request.onerror = () => reject(request.error ?? new Error('Could not open local artifact storage'))
  })
  try {
    await new Promise<void>((resolve, reject) => {
      const transaction = db.transaction('artifacts', 'readwrite')
      const artifacts = transaction.objectStore('artifacts')
      const existing = artifacts.get(key)
      let conflict = false
      existing.onsuccess = () => {
        const previous = existing.result as { sha256?: string } | undefined
        if (previous && previous.sha256 !== value.sha256) { conflict = true; transaction.abort(); return }
        artifacts.put(value, key)
      }
      transaction.oncomplete = () => resolve()
      transaction.onerror = () => reject(transaction.error ?? new Error('Could not persist received artifact'))
      transaction.onabort = () => reject(new Error(conflict ? 'Artifact names are immutable; use a new name for changed bytes' : 'Artifact persistence was aborted'))
    })
  } finally { db.close() }
}

async function loadArtifact(name: string): Promise<Uint8Array | null> {
  const db = await new Promise<IDBDatabase>((resolve, reject) => {
    const request = indexedDB.open('mesh-peer-artifacts', 1)
    request.onupgradeneeded = () => request.result.createObjectStore('artifacts')
    request.onsuccess = () => resolve(request.result)
    request.onerror = () => reject(request.error ?? new Error('Could not open local artifact storage'))
  })
  try {
    return await new Promise<Uint8Array | null>((resolve, reject) => {
      const request = db.transaction('artifacts', 'readonly').objectStore('artifacts').get(`artifact:${name}`)
      request.onsuccess = () => resolve((request.result as { bytes?: Uint8Array } | undefined)?.bytes ?? null)
      request.onerror = () => reject(request.error ?? new Error('Could not read local artifact storage'))
    })
  } finally { db.close() }
}

export class PeerMeshNode {
  private peers = new Map<string, RTCPeerConnection>()
  private channels = new Map<string, RTCDataChannel>()
  private incoming = new Map<string, IncomingTransfer>()
  private pending = new Map<string, PendingTransfer>()
  private pendingArtifactReads = new Map<string, PendingArtifactRead>()
  private signalQueues = new Map<string, Promise<void>>()
  private earlyCandidates = new WeakMap<RTCPeerConnection, RTCIceCandidateInit[]>()
  private closed = false

  constructor(
    private readonly nodeId: string,
    private readonly signal: (peerNodeId: string, payload: SignalPayload) => void,
    private readonly onNotice: (message: string) => void,
    private readonly onStored?: (name: string, sha256: string, bytes: number) => Promise<void>,
  ) {}

  addPeers(peerNodeIds: string[]) {
    for (const peerId of peerNodeIds) this.considerPeer(peerId)
  }

  canTransferTo(peerId: string) {
    const peer = this.peers.get(peerId)
    return Boolean(peer && peer.signalingState !== 'closed' && peer.connectionState !== 'failed' && peer.connectionState !== 'closed')
  }

  peerJoined(peerId: string) {
    if (this.peers.has(peerId)) this.peerLeft(peerId)
    this.considerPeer(peerId)
  }

  peerLeft(peerId: string) {
    const oldPeer = this.peers.get(peerId)
    this.peers.delete(peerId)
    this.channels.delete(peerId)
    this.signalQueues.delete(peerId)
    oldPeer?.close()
    for (const [transferId, transfer] of this.incoming) if (transfer.peerId === peerId) this.incoming.delete(transferId)
    for (const [transferId, pending] of this.pending) if (pending.peerId === peerId) {
      window.clearTimeout(pending.timer)
      this.pending.delete(transferId)
      pending.reject(new Error('Peer disconnected during artifact transfer'))
    }
    for (const [key, pending] of this.pendingArtifactReads) if (pending.peerId === peerId) {
      window.clearTimeout(pending.timer)
      this.pendingArtifactReads.delete(key)
      pending.reject(new Error('Peer disconnected while a shard was being requested'))
    }
  }

  private considerPeer(peerId: string) {
    if (peerId === this.nodeId || this.closed || this.peers.has(peerId)) return
    const initiates = this.nodeId.localeCompare(peerId) < 0
    const peer = this.createPeer(peerId, initiates)
    if (initiates) void peer.createOffer().then(async offer => {
      await peer.setLocalDescription(offer)
      this.signal(peerId, { description: peer.localDescription ?? offer })
    }).catch(() => this.onNotice('Could not negotiate a direct peer connection'))
  }

  async handleSignal(peerId: string, payload: SignalPayload) {
    if (this.closed || peerId === this.nodeId) return
    const peer = this.peers.get(peerId) ?? this.createPeer(peerId, false)
    const previous = this.signalQueues.get(peerId) ?? Promise.resolve()
    const next = previous.then(async () => {
      if (this.closed || this.peers.get(peerId) !== peer) return
      if (payload.description?.type === 'offer') {
        await peer.setRemoteDescription(payload.description)
        await this.flushCandidates(peer)
        const answer = await peer.createAnswer()
        await peer.setLocalDescription(answer)
        this.signal(peerId, { description: peer.localDescription ?? answer })
      } else if (payload.description?.type === 'answer') {
        await peer.setRemoteDescription(payload.description)
        await this.flushCandidates(peer)
      } else if (payload.candidate) {
        if (peer.remoteDescription) await peer.addIceCandidate(payload.candidate)
        else {
          const candidates = this.earlyCandidates.get(peer) ?? []
          if (candidates.length >= 128) throw new Error('Too many early ICE candidates')
          candidates.push(payload.candidate)
          this.earlyCandidates.set(peer, candidates)
        }
      }
    }).catch(() => {
      if (!this.closed && this.peers.get(peerId) === peer) this.onNotice('Peer negotiation failed; the node will retry when it reconnects')
    })
    this.signalQueues.set(peerId, next)
    await next
    if (this.signalQueues.get(peerId) === next) this.signalQueues.delete(peerId)
  }

  private async flushCandidates(peer: RTCPeerConnection) {
    const candidates = this.earlyCandidates.get(peer) ?? []
    this.earlyCandidates.delete(peer)
    for (const candidate of candidates) await peer.addIceCandidate(candidate)
  }

  private createPeer(peerId: string, initiator: boolean) {
    // Browser E2E runs three peer pages on the same host, so local host
    // candidates are enough and avoid making CI depend on an external STUN
    // service. Production keeps the normal STUN-assisted ICE configuration.
    const localE2e = new URLSearchParams(location.search).get('mesh-e2e') === 'local'
    const peer = new RTCPeerConnection({ iceServers: localE2e ? [] : [{ urls: 'stun:stun.l.google.com:19302' }] })
    this.peers.set(peerId, peer)
    peer.onicecandidate = event => { if (event.candidate) this.signal(peerId, { candidate: event.candidate.toJSON() }) }
    peer.onconnectionstatechange = () => {
      if ((peer.connectionState === 'failed' || peer.connectionState === 'closed') && this.peers.get(peerId) === peer) this.peerLeft(peerId)
    }
    peer.ondatachannel = event => this.bindChannel(peerId, event.channel)
    if (initiator) this.bindChannel(peerId, peer.createDataChannel('mesh-artifacts', { ordered: true }))
    return peer
  }

  private bindChannel(peerId: string, channel: RTCDataChannel) {
    channel.binaryType = 'arraybuffer'
    channel.bufferedAmountLowThreshold = 64 * 1024
    this.channels.set(peerId, channel)
    channel.onmessage = event => { void this.handleDataMessage(peerId, channel, event.data) }
    channel.onclose = () => { if (this.channels.get(peerId) === channel) this.channels.delete(peerId) }
  }

  private async handleDataMessage(peerId: string, channel: RTCDataChannel, data: unknown) {
    if (typeof data !== 'string' || data.length > MAX_FRAME_CHARS) return
    // Failed senders can leave an incomplete frame sequence on a live channel.
    // Reclaim idle slots before accepting fresh traffic, keeping retries usable.
    const at = performance.now()
    for (const [id, transfer] of this.incoming) if (at - transfer.lastActivity > 45_000) this.incoming.delete(id)
    let message: Record<string, unknown>
    try { message = JSON.parse(data) as Record<string, unknown> } catch { return }
    if (message.type === 'artifact_ack' && typeof message.transfer_id === 'string' && typeof message.sha256 === 'string') {
      const pending = this.pending.get(message.transfer_id)
      if (pending?.peerId === peerId) { window.clearTimeout(pending.timer); this.pending.delete(message.transfer_id); pending.resolve(message.sha256) }
      return
    }
    if (message.type === 'artifact_get' && typeof message.name === 'string' && message.name.length > 0 && message.name.length <= 128) {
      try {
        const artifact = await loadArtifact(message.name)
        if (!artifact) { channel.send(JSON.stringify({ type: 'artifact_missing', name: message.name })); return }
        void this.sendArtifact(peerId, message.name, artifact, false).catch(() => undefined)
      } catch { channel.send(JSON.stringify({ type: 'artifact_missing', name: message.name })) }
      return
    }
    if (message.type === 'artifact_missing' && typeof message.name === 'string') {
      const key = `${peerId}:${message.name}`
      const pending = this.pendingArtifactReads.get(key)
      if (pending) { window.clearTimeout(pending.timer); this.pendingArtifactReads.delete(key); pending.reject(new Error(`Peer ${peerId} no longer has shard “${message.name}”`)) }
      return
    }
    if (message.type === 'artifact_start') {
      const start = message as unknown as TransferStart
      if (typeof start.transfer_id !== 'string' || start.transfer_id.length > 64 || !start.transfer_id || typeof start.name !== 'string' || !start.name || start.name.length > 128 || !Number.isSafeInteger(start.total_bytes) || start.total_bytes <= 0 || start.total_bytes > MAX_ARTIFACT_BYTES
        || !Number.isSafeInteger(start.chunks) || start.chunks < 0 || start.chunks > Math.ceil(MAX_ARTIFACT_BYTES / CHUNK_BYTES)
        || !/^[a-f0-9]{64}$/i.test(start.sha256) || start.chunks !== Math.ceil(start.total_bytes / CHUNK_BYTES)) return
      if (this.incoming.size >= MAX_CONCURRENT_TRANSFERS && !this.incoming.has(start.transfer_id)) {
        channel.send(JSON.stringify({ type: 'artifact_nack', transfer_id: start.transfer_id, reason: 'receiver_busy' }))
        return
      }
      this.incoming.set(start.transfer_id, { peerId, start, chunks: new Map(), bytes: 0, lastActivity: at })
      return
    }
    if (message.type === 'artifact_chunk' && typeof message.transfer_id === 'string' && typeof message.data === 'string') {
      const index = Number(message.index)
      if (!Number.isSafeInteger(index)) return
      const transfer = this.incoming.get(message.transfer_id)
      if (!transfer || index < 0 || index >= transfer.start.chunks || transfer.chunks.has(index)) return
      let bytes: Uint8Array
      try { bytes = decode(message.data) } catch { this.incoming.delete(message.transfer_id); return }
      if (bytes.length > CHUNK_BYTES || transfer.bytes + bytes.length > transfer.start.total_bytes) { this.incoming.delete(message.transfer_id); return }
      transfer.chunks.set(index, bytes)
      transfer.bytes += bytes.length
      transfer.lastActivity = at
      if (transfer.chunks.size !== transfer.start.chunks) return
      const result = new Uint8Array(transfer.bytes)
      let offset = 0
      for (let index = 0; index < transfer.start.chunks; index++) {
        const chunk = transfer.chunks.get(index)
        if (!chunk) { this.incoming.delete(message.transfer_id); return }
        result.set(chunk, offset); offset += chunk.length
      }
      const actualHash = await digest(result)
      if (result.length !== transfer.start.total_bytes || actualHash !== transfer.start.sha256.toLowerCase()) {
        this.incoming.delete(message.transfer_id)
        channel.send(JSON.stringify({ type: 'artifact_nack', transfer_id: message.transfer_id, reason: 'hash_mismatch' }))
        return
      }
      try {
        await persistArtifact(`artifact:${transfer.start.name}`, { name: transfer.start.name, bytes: result, sha256: actualHash, receivedAt: Date.now() })
        await this.onStored?.(transfer.start.name, actualHash, result.length)
        channel.send(JSON.stringify({ type: 'artifact_ack', transfer_id: message.transfer_id, sha256: actualHash }))
        this.onNotice(`Verified peer artifact “${transfer.start.name}” (${result.length.toLocaleString()} bytes)`)
        const readKey = `${peerId}:${transfer.start.name}`
        const pendingRead = this.pendingArtifactReads.get(readKey)
        if (pendingRead) { window.clearTimeout(pendingRead.timer); this.pendingArtifactReads.delete(readKey); pendingRead.resolve(result) }
      } catch { channel.send(JSON.stringify({ type: 'artifact_nack', transfer_id: message.transfer_id, reason: 'storage_failed' })) }
      this.incoming.delete(message.transfer_id)
    }
    if (message.type === 'artifact_nack' && typeof message.transfer_id === 'string') {
      const pending = this.pending.get(message.transfer_id)
      if (pending?.peerId === peerId) { window.clearTimeout(pending.timer); this.pending.delete(message.transfer_id); pending.reject(new Error(message.reason === 'receiver_busy' ? 'Peer storage is busy; retry the artifact transfer' : 'Peer rejected the artifact: integrity check or storage failed')) }
    }
  }

  async sendArtifact(peerId: string, name: string, bytes: Uint8Array, persistLocally = true) {
    if (this.closed) throw new Error('Node connection closed')
    if (!name || name.length > 128) throw new Error('Peer artifact names must contain 1–128 characters')
    if (!bytes.length) throw new Error('Peer artifacts must contain at least one byte')
    if (bytes.length > MAX_ARTIFACT_BYTES) throw new Error('Peer artifact exceeds the 8 MiB transfer limit')
    this.onNotice(`Sending checkpoint artifact “${name}” to peer ${peerId}…`)
    // Give the local end-to-end harness a window to inject a peer failure while
    // a transfer is in flight. Production transfers are not delayed.
    if (new URLSearchParams(location.search).get('mesh-e2e') === 'local') await new Promise(resolve => window.setTimeout(resolve, 750))
    const deadline = Date.now() + 20_000
    let channel = this.channels.get(peerId)
    while ((!channel || channel.readyState !== 'open') && Date.now() < deadline) {
      if (this.closed) throw new Error('Node connection closed')
      await new Promise(resolve => window.setTimeout(resolve, 100))
      channel = this.channels.get(peerId)
    }
    if (!channel || channel.readyState !== 'open') throw new Error('Direct peer channel did not become ready; the network may block peer connections')
    const transferId = crypto.randomUUID()
    const sha256 = await digest(bytes)
    const chunks = Math.ceil(bytes.length / CHUNK_BYTES)
    const acknowledged = new Promise<string>((resolve, reject) => {
      const timer = window.setTimeout(() => { this.pending.delete(transferId); reject(new Error('Peer did not confirm the artifact before timeout')) }, 30_000)
      this.pending.set(transferId, { peerId, resolve, reject, timer })
    })
    // Disconnect can reject the acknowledgement while chunk sends are awaiting
    // backpressure. Observe it immediately, then propagate it at the await below.
    void acknowledged.catch(() => undefined)
    try {
      await this.sendFrame(channel, { type: 'artifact_start', transfer_id: transferId, name, total_bytes: bytes.length, chunks, sha256 } satisfies TransferStart)
      for (let index = 0; index < chunks; index++) {
        const part = bytes.subarray(index * CHUNK_BYTES, Math.min(bytes.length, (index + 1) * CHUNK_BYTES))
        await this.sendFrame(channel, { type: 'artifact_chunk', transfer_id: transferId, index, data: encode(part) } satisfies TransferChunk)
      }
    } catch (error) {
      const pending = this.pending.get(transferId)
      if (pending) { window.clearTimeout(pending.timer); this.pending.delete(transferId) }
      throw error
    }
    const receivedHash = await acknowledged
    if (receivedHash !== sha256) throw new Error('Peer acknowledged a different artifact digest')
    if (persistLocally) {
      await persistArtifact(`artifact:${name}`, { name, bytes: bytes.slice(), sha256, receivedAt: Date.now() })
      await this.onStored?.(name, sha256, bytes.length)
    }
    return { transferId, sha256, bytes: bytes.length }
  }

  async storeArtifact(name: string, bytes: Uint8Array) {
    if (!name || name.length > 128 || !bytes.length || bytes.length > MAX_ARTIFACT_BYTES) throw new Error('Local shard is outside artifact limits')
    const sha256 = await digest(bytes)
    await persistArtifact(`artifact:${name}`, { name, bytes: bytes.slice(), sha256, receivedAt: Date.now() })
    await this.onStored?.(name, sha256, bytes.length)
    return { sha256, bytes: bytes.length }
  }

  async getArtifact(peerId: string, name: string) {
    const local = await loadArtifact(name)
    if (local) return local
    if (peerId === this.nodeId) throw new Error(`This node no longer has shard “${name}”`)
    const deadline = Date.now() + 20_000
    let channel = this.channels.get(peerId)
    while ((!channel || channel.readyState !== 'open') && Date.now() < deadline) {
      await new Promise(resolve => window.setTimeout(resolve, 100))
      channel = this.channels.get(peerId)
    }
    if (!channel || channel.readyState !== 'open') throw new Error(`Peer ${peerId} is not connected`)
    const key = `${peerId}:${name}`
    if (this.pendingArtifactReads.has(key)) throw new Error('A request for this shard is already pending')
    const result = new Promise<Uint8Array>((resolve, reject) => {
      const timer = window.setTimeout(() => {
        this.pendingArtifactReads.delete(key)
        reject(new Error(`Timed out requesting shard “${name}” from peer ${peerId}`))
      }, 30_000)
      this.pendingArtifactReads.set(key, { peerId, name, resolve, reject, timer })
    })
    try { await this.sendFrame(channel, { type: 'artifact_get', name }) }
    catch (error) {
      const pending = this.pendingArtifactReads.get(key)
      if (pending) { window.clearTimeout(pending.timer); this.pendingArtifactReads.delete(key) }
      throw error
    }
    return result
  }

  private async sendFrame(channel: RTCDataChannel, message: unknown) {
    const frame = JSON.stringify(message)
    if (frame.length > MAX_FRAME_CHARS) throw new Error('Peer transfer frame exceeds its size limit')
    while (channel.bufferedAmount > BUFFER_LIMIT_BYTES) {
      await new Promise<void>((resolve, reject) => {
        const timeout = window.setTimeout(() => reject(new Error('Peer data channel remained congested')), 15_000)
        channel.addEventListener('bufferedamountlow', () => { window.clearTimeout(timeout); resolve() }, { once: true })
        if (channel.readyState !== 'open') { window.clearTimeout(timeout); reject(new Error('Peer disconnected during transfer')) }
      })
    }
    if (channel.readyState !== 'open') throw new Error('Peer disconnected during transfer')
    channel.send(frame)
  }

  close() {
    this.closed = true
    for (const peer of this.peers.values()) peer.close()
    this.peers.clear(); this.channels.clear(); this.incoming.clear()
    this.signalQueues.clear()
    for (const pending of this.pending.values()) { window.clearTimeout(pending.timer); pending.reject(new Error('Node connection closed')) }
    this.pending.clear()
    for (const pending of this.pendingArtifactReads.values()) {
      window.clearTimeout(pending.timer)
      pending.reject(new Error('Node connection closed while reading a shard'))
    }
    this.pendingArtifactReads.clear()
  }
}

	const MAX_ARTIFACT_BYTES = 8 * 1024 * 1024
	const CHUNK_BYTES = 16 * 1024
	const MAX_FRAME_CHARS = 24 * 1024
	const BUFFER_LIMIT_BYTES = 256 * 1024
	const MAX_CONCURRENT_TRANSFERS = 2

	type SignalPayload = { description?: RTCSessionDescriptionInit; candidate?: RTCIceCandidateInit }
	type TransferStart = { type: 'artifact_start'; transfer_id: string; name: string; total_bytes: number; chunks: number; sha256: string }
	type TransferChunk = { type: 'artifact_chunk'; transfer_id: string; index: number; data: string }
	type IncomingTransfer = { peerId: string; start: TransferStart; chunks: Map<number, Uint8Array>; bytes: number; lastActivity: number }
	type PendingTransfer = { peerId: string; resolve: (digest: string) => void; reject: (error: Error) => void; timer: number }
	type PendingArtifactRead = { peerId: string; name: string; resolve: (bytes: Uint8Array) => void; reject: (error: Error) => void; timer: number }

	function encode(bytes: Uint8Array) {
	let binary = ''
	for (let offset = 0; offset < bytes.length; offset += 0x8000) binary += String.fromCharCode(...bytes.subarray(offset, offset + 0x8000))
	return btoa(binary)
	}

	function decode(base64: string): Uint8Array {
	const binary = atob(base64)
	const result = new Uint8Array(binary.length)
	for (let i = 0; i < binary.length; i++) result[i] = binary.charCodeAt(i)
	return result
	}

	async function digest(bytes: Uint8Array) {
	const copy = bytes.slice().buffer
	const result = await crypto.subtle.digest('SHA-256', copy)
	return Array.from(new Uint8Array(result), byte => byte.toString(16).padStart(2, '0')).join('')
	}

	async function persistArtifact(key: string, value: { name: string; bytes: Uint8Array; sha256: string; receivedAt: number }) {
	const db = await new Promise<IDBDatabase>((resolve, reject) => {
	const request = indexedDB.open('mesh-peer-artifacts', 1)
	request.onupgradeneeded = () => request.result.createObjectStore('artifacts')
	request.onsuccess = () => resolve(request.result)
	request.onerror = () => reject(request.error ?? new Error('Could not open local artifact storage'))
	})
	try {
	await new Promise<void>((resolve, reject) => {
	const transaction = db.transaction('artifacts', 'readwrite')
	const artifacts = transaction.objectStore('artifacts')
	const existing = artifacts.get(key)
	let conflict = false
	existing.onsuccess = () => {
	const previous = existing.result as { sha256?: string } \| undefined
	if (previous && previous.sha256 !== value.sha256) { conflict = true; transaction.abort(); return }
	artifacts.put(value, key)
	}
	transaction.oncomplete = () => resolve()
	transaction.onerror = () => reject(transaction.error ?? new Error('Could not persist received artifact'))
	transaction.onabort = () => reject(new Error(conflict ? 'Artifact names are immutable; use a new name for changed bytes' : 'Artifact persistence was aborted'))
	})
	} finally { db.close() }
	}

	async function loadArtifact(name: string): Promise<Uint8Array \| null> {
	const db = await new Promise<IDBDatabase>((resolve, reject) => {
	const request = indexedDB.open('mesh-peer-artifacts', 1)
	request.onupgradeneeded = () => request.result.createObjectStore('artifacts')
	request.onsuccess = () => resolve(request.result)
	request.onerror = () => reject(request.error ?? new Error('Could not open local artifact storage'))
	})
	try {
	return await new Promise<Uint8Array \| null>((resolve, reject) => {
	const request = db.transaction('artifacts', 'readonly').objectStore('artifacts').get(`artifact:${name}`)
	request.onsuccess = () => resolve((request.result as { bytes?: Uint8Array } \| undefined)?.bytes ?? null)
	request.onerror = () => reject(request.error ?? new Error('Could not read local artifact storage'))
	})
	} finally { db.close() }
	}

	export class PeerMeshNode {
	private peers = new Map<string, RTCPeerConnection>()
	private channels = new Map<string, RTCDataChannel>()
	private incoming = new Map<string, IncomingTransfer>()
	private pending = new Map<string, PendingTransfer>()
	private pendingArtifactReads = new Map<string, PendingArtifactRead>()
	private signalQueues = new Map<string, Promise<void>>()
	private earlyCandidates = new WeakMap<RTCPeerConnection, RTCIceCandidateInit[]>()
	private closed = false

	constructor(
	private readonly nodeId: string,
	private readonly signal: (peerNodeId: string, payload: SignalPayload) => void,
	private readonly onNotice: (message: string) => void,
	private readonly onStored?: (name: string, sha256: string, bytes: number) => Promise<void>,
	) {}

	addPeers(peerNodeIds: string[]) {
	for (const peerId of peerNodeIds) this.considerPeer(peerId)
	}

	canTransferTo(peerId: string) {
	const peer = this.peers.get(peerId)
	return Boolean(peer && peer.signalingState !== 'closed' && peer.connectionState !== 'failed' && peer.connectionState !== 'closed')
	}

	peerJoined(peerId: string) {
	if (this.peers.has(peerId)) this.peerLeft(peerId)
	this.considerPeer(peerId)
	}

	peerLeft(peerId: string) {
	const oldPeer = this.peers.get(peerId)
	this.peers.delete(peerId)
	this.channels.delete(peerId)
	this.signalQueues.delete(peerId)
	oldPeer?.close()
	for (const [transferId, transfer] of this.incoming) if (transfer.peerId === peerId) this.incoming.delete(transferId)
	for (const [transferId, pending] of this.pending) if (pending.peerId === peerId) {
	window.clearTimeout(pending.timer)
	this.pending.delete(transferId)
	pending.reject(new Error('Peer disconnected during artifact transfer'))
	}
	for (const [key, pending] of this.pendingArtifactReads) if (pending.peerId === peerId) {
	window.clearTimeout(pending.timer)
	this.pendingArtifactReads.delete(key)
	pending.reject(new Error('Peer disconnected while a shard was being requested'))
	}
	}

	private considerPeer(peerId: string) {
	if (peerId === this.nodeId \|\| this.closed \|\| this.peers.has(peerId)) return
	const initiates = this.nodeId.localeCompare(peerId) < 0
	const peer = this.createPeer(peerId, initiates)
	if (initiates) void peer.createOffer().then(async offer => {
	await peer.setLocalDescription(offer)
	this.signal(peerId, { description: peer.localDescription ?? offer })
	}).catch(() => this.onNotice('Could not negotiate a direct peer connection'))
	}

	async handleSignal(peerId: string, payload: SignalPayload) {
	if (this.closed \|\| peerId === this.nodeId) return
	const peer = this.peers.get(peerId) ?? this.createPeer(peerId, false)
	const previous = this.signalQueues.get(peerId) ?? Promise.resolve()
	const next = previous.then(async () => {
	if (this.closed \|\| this.peers.get(peerId) !== peer) return
	if (payload.description?.type === 'offer') {
	await peer.setRemoteDescription(payload.description)
	await this.flushCandidates(peer)
	const answer = await peer.createAnswer()
	await peer.setLocalDescription(answer)
	this.signal(peerId, { description: peer.localDescription ?? answer })
	} else if (payload.description?.type === 'answer') {
	await peer.setRemoteDescription(payload.description)
	await this.flushCandidates(peer)
	} else if (payload.candidate) {
	if (peer.remoteDescription) await peer.addIceCandidate(payload.candidate)
	else {
	const candidates = this.earlyCandidates.get(peer) ?? []
	if (candidates.length >= 128) throw new Error('Too many early ICE candidates')
	candidates.push(payload.candidate)
	this.earlyCandidates.set(peer, candidates)
	}
	}
	}).catch(() => {
	if (!this.closed && this.peers.get(peerId) === peer) this.onNotice('Peer negotiation failed; the node will retry when it reconnects')
	})
	this.signalQueues.set(peerId, next)
	await next
	if (this.signalQueues.get(peerId) === next) this.signalQueues.delete(peerId)
	}

	private async flushCandidates(peer: RTCPeerConnection) {
	const candidates = this.earlyCandidates.get(peer) ?? []
	this.earlyCandidates.delete(peer)
	for (const candidate of candidates) await peer.addIceCandidate(candidate)
	}

	private createPeer(peerId: string, initiator: boolean) {
	// Browser E2E runs three peer pages on the same host, so local host
	// candidates are enough and avoid making CI depend on an external STUN
	// service. Production keeps the normal STUN-assisted ICE configuration.
	const localE2e = new URLSearchParams(location.search).get('mesh-e2e') === 'local'
	const peer = new RTCPeerConnection({ iceServers: localE2e ? [] : [{ urls: 'stun:stun.l.google.com:19302' }] })
	this.peers.set(peerId, peer)
	peer.onicecandidate = event => { if (event.candidate) this.signal(peerId, { candidate: event.candidate.toJSON() }) }
	peer.onconnectionstatechange = () => {
	if ((peer.connectionState === 'failed' \|\| peer.connectionState === 'closed') && this.peers.get(peerId) === peer) this.peerLeft(peerId)
	}
	peer.ondatachannel = event => this.bindChannel(peerId, event.channel)
	if (initiator) this.bindChannel(peerId, peer.createDataChannel('mesh-artifacts', { ordered: true }))
	return peer
	}

	private bindChannel(peerId: string, channel: RTCDataChannel) {
	channel.binaryType = 'arraybuffer'
	channel.bufferedAmountLowThreshold = 64 * 1024
	this.channels.set(peerId, channel)
	channel.onmessage = event => { void this.handleDataMessage(peerId, channel, event.data) }
	channel.onclose = () => { if (this.channels.get(peerId) === channel) this.channels.delete(peerId) }
	}

	private async handleDataMessage(peerId: string, channel: RTCDataChannel, data: unknown) {
	if (typeof data !== 'string' \|\| data.length > MAX_FRAME_CHARS) return
	// Failed senders can leave an incomplete frame sequence on a live channel.
	// Reclaim idle slots before accepting fresh traffic, keeping retries usable.
	const at = performance.now()
	for (const [id, transfer] of this.incoming) if (at - transfer.lastActivity > 45_000) this.incoming.delete(id)
	let message: Record<string, unknown>
	try { message = JSON.parse(data) as Record<string, unknown> } catch { return }
	if (message.type === 'artifact_ack' && typeof message.transfer_id === 'string' && typeof message.sha256 === 'string') {
	const pending = this.pending.get(message.transfer_id)
	if (pending?.peerId === peerId) { window.clearTimeout(pending.timer); this.pending.delete(message.transfer_id); pending.resolve(message.sha256) }
	return
	}
	if (message.type === 'artifact_get' && typeof message.name === 'string' && message.name.length > 0 && message.name.length <= 128) {
	try {
	const artifact = await loadArtifact(message.name)
	if (!artifact) { channel.send(JSON.stringify({ type: 'artifact_missing', name: message.name })); return }
	void this.sendArtifact(peerId, message.name, artifact, false).catch(() => undefined)
	} catch { channel.send(JSON.stringify({ type: 'artifact_missing', name: message.name })) }
	return
	}
	if (message.type === 'artifact_missing' && typeof message.name === 'string') {
	const key = `${peerId}:${message.name}`
	const pending = this.pendingArtifactReads.get(key)
	if (pending) { window.clearTimeout(pending.timer); this.pendingArtifactReads.delete(key); pending.reject(new Error(`Peer ${peerId} no longer has shard “${message.name}”`)) }
	return
	}
	if (message.type === 'artifact_start') {
	const start = message as unknown as TransferStart
	if (typeof start.transfer_id !== 'string' \|\| start.transfer_id.length > 64 \|\| !start.transfer_id \|\| typeof start.name !== 'string' \|\| !start.name \|\| start.name.length > 128 \|\| !Number.isSafeInteger(start.total_bytes) \|\| start.total_bytes <= 0 \|\| start.total_bytes > MAX_ARTIFACT_BYTES
	\|\| !Number.isSafeInteger(start.chunks) \|\| start.chunks < 0 \|\| start.chunks > Math.ceil(MAX_ARTIFACT_BYTES / CHUNK_BYTES)
	\|\| !/^[a-f0-9]{64}$/i.test(start.sha256) \|\| start.chunks !== Math.ceil(start.total_bytes / CHUNK_BYTES)) return
	if (this.incoming.size >= MAX_CONCURRENT_TRANSFERS && !this.incoming.has(start.transfer_id)) {
	channel.send(JSON.stringify({ type: 'artifact_nack', transfer_id: start.transfer_id, reason: 'receiver_busy' }))
	return
	}
	this.incoming.set(start.transfer_id, { peerId, start, chunks: new Map(), bytes: 0, lastActivity: at })
	return
	}
	if (message.type === 'artifact_chunk' && typeof message.transfer_id === 'string' && typeof message.data === 'string') {
	const index = Number(message.index)
	if (!Number.isSafeInteger(index)) return
	const transfer = this.incoming.get(message.transfer_id)
	if (!transfer \|\| index < 0 \|\| index >= transfer.start.chunks \|\| transfer.chunks.has(index)) return
	let bytes: Uint8Array
	try { bytes = decode(message.data) } catch { this.incoming.delete(message.transfer_id); return }
	if (bytes.length > CHUNK_BYTES \|\| transfer.bytes + bytes.length > transfer.start.total_bytes) { this.incoming.delete(message.transfer_id); return }
	transfer.chunks.set(index, bytes)
	transfer.bytes += bytes.length
	transfer.lastActivity = at
	if (transfer.chunks.size !== transfer.start.chunks) return
	const result = new Uint8Array(transfer.bytes)
	let offset = 0
	for (let index = 0; index < transfer.start.chunks; index++) {
	const chunk = transfer.chunks.get(index)
	if (!chunk) { this.incoming.delete(message.transfer_id); return }
	result.set(chunk, offset); offset += chunk.length
	}
	const actualHash = await digest(result)
	if (result.length !== transfer.start.total_bytes \|\| actualHash !== transfer.start.sha256.toLowerCase()) {
	this.incoming.delete(message.transfer_id)
	channel.send(JSON.stringify({ type: 'artifact_nack', transfer_id: message.transfer_id, reason: 'hash_mismatch' }))
	return
	}
	try {
	await persistArtifact(`artifact:${transfer.start.name}`, { name: transfer.start.name, bytes: result, sha256: actualHash, receivedAt: Date.now() })
	await this.onStored?.(transfer.start.name, actualHash, result.length)
	channel.send(JSON.stringify({ type: 'artifact_ack', transfer_id: message.transfer_id, sha256: actualHash }))
	this.onNotice(`Verified peer artifact “${transfer.start.name}” (${result.length.toLocaleString()} bytes)`)
	const readKey = `${peerId}:${transfer.start.name}`
	const pendingRead = this.pendingArtifactReads.get(readKey)
	if (pendingRead) { window.clearTimeout(pendingRead.timer); this.pendingArtifactReads.delete(readKey); pendingRead.resolve(result) }
	} catch { channel.send(JSON.stringify({ type: 'artifact_nack', transfer_id: message.transfer_id, reason: 'storage_failed' })) }
	this.incoming.delete(message.transfer_id)
	}
	if (message.type === 'artifact_nack' && typeof message.transfer_id === 'string') {
	const pending = this.pending.get(message.transfer_id)
	if (pending?.peerId === peerId) { window.clearTimeout(pending.timer); this.pending.delete(message.transfer_id); pending.reject(new Error(message.reason === 'receiver_busy' ? 'Peer storage is busy; retry the artifact transfer' : 'Peer rejected the artifact: integrity check or storage failed')) }
	}
	}

	async sendArtifact(peerId: string, name: string, bytes: Uint8Array, persistLocally = true) {
	if (this.closed) throw new Error('Node connection closed')
	if (!name \|\| name.length > 128) throw new Error('Peer artifact names must contain 1–128 characters')
	if (!bytes.length) throw new Error('Peer artifacts must contain at least one byte')
	if (bytes.length > MAX_ARTIFACT_BYTES) throw new Error('Peer artifact exceeds the 8 MiB transfer limit')
	this.onNotice(`Sending checkpoint artifact “${name}” to peer ${peerId}…`)
	// Give the local end-to-end harness a window to inject a peer failure while
	// a transfer is in flight. Production transfers are not delayed.
	if (new URLSearchParams(location.search).get('mesh-e2e') === 'local') await new Promise(resolve => window.setTimeout(resolve, 750))
	const deadline = Date.now() + 20_000
	let channel = this.channels.get(peerId)
	while ((!channel \|\| channel.readyState !== 'open') && Date.now() < deadline) {
	if (this.closed) throw new Error('Node connection closed')
	await new Promise(resolve => window.setTimeout(resolve, 100))
	channel = this.channels.get(peerId)
	}
	if (!channel \|\| channel.readyState !== 'open') throw new Error('Direct peer channel did not become ready; the network may block peer connections')
	const transferId = crypto.randomUUID()
	const sha256 = await digest(bytes)
	const chunks = Math.ceil(bytes.length / CHUNK_BYTES)
	const acknowledged = new Promise<string>((resolve, reject) => {
	const timer = window.setTimeout(() => { this.pending.delete(transferId); reject(new Error('Peer did not confirm the artifact before timeout')) }, 30_000)
	this.pending.set(transferId, { peerId, resolve, reject, timer })
	})
	// Disconnect can reject the acknowledgement while chunk sends are awaiting
	// backpressure. Observe it immediately, then propagate it at the await below.
	void acknowledged.catch(() => undefined)
	try {
	await this.sendFrame(channel, { type: 'artifact_start', transfer_id: transferId, name, total_bytes: bytes.length, chunks, sha256 } satisfies TransferStart)
	for (let index = 0; index < chunks; index++) {
	const part = bytes.subarray(index * CHUNK_BYTES, Math.min(bytes.length, (index + 1) * CHUNK_BYTES))
	await this.sendFrame(channel, { type: 'artifact_chunk', transfer_id: transferId, index, data: encode(part) } satisfies TransferChunk)
	}
	} catch (error) {
	const pending = this.pending.get(transferId)
	if (pending) { window.clearTimeout(pending.timer); this.pending.delete(transferId) }
	throw error
	}
	const receivedHash = await acknowledged
	if (receivedHash !== sha256) throw new Error('Peer acknowledged a different artifact digest')
	if (persistLocally) {
	await persistArtifact(`artifact:${name}`, { name, bytes: bytes.slice(), sha256, receivedAt: Date.now() })
	await this.onStored?.(name, sha256, bytes.length)
	}
	return { transferId, sha256, bytes: bytes.length }
	}

	async storeArtifact(name: string, bytes: Uint8Array) {
	if (!name \|\| name.length > 128 \|\| !bytes.length \|\| bytes.length > MAX_ARTIFACT_BYTES) throw new Error('Local shard is outside artifact limits')
	const sha256 = await digest(bytes)
	await persistArtifact(`artifact:${name}`, { name, bytes: bytes.slice(), sha256, receivedAt: Date.now() })
	await this.onStored?.(name, sha256, bytes.length)
	return { sha256, bytes: bytes.length }
	}

	async getArtifact(peerId: string, name: string) {
	const local = await loadArtifact(name)
	if (local) return local
	if (peerId === this.nodeId) throw new Error(`This node no longer has shard “${name}”`)
	const deadline = Date.now() + 20_000
	let channel = this.channels.get(peerId)
	while ((!channel \|\| channel.readyState !== 'open') && Date.now() < deadline) {
	await new Promise(resolve => window.setTimeout(resolve, 100))
	channel = this.channels.get(peerId)
	}
	if (!channel \|\| channel.readyState !== 'open') throw new Error(`Peer ${peerId} is not connected`)
	const key = `${peerId}:${name}`
	if (this.pendingArtifactReads.has(key)) throw new Error('A request for this shard is already pending')
	const result = new Promise<Uint8Array>((resolve, reject) => {
	const timer = window.setTimeout(() => {
	this.pendingArtifactReads.delete(key)
	reject(new Error(`Timed out requesting shard “${name}” from peer ${peerId}`))
	}, 30_000)
	this.pendingArtifactReads.set(key, { peerId, name, resolve, reject, timer })
	})
	try { await this.sendFrame(channel, { type: 'artifact_get', name }) }
	catch (error) {
	const pending = this.pendingArtifactReads.get(key)
	if (pending) { window.clearTimeout(pending.timer); this.pendingArtifactReads.delete(key) }
	throw error
	}
	return result
	}

	private async sendFrame(channel: RTCDataChannel, message: unknown) {
	const frame = JSON.stringify(message)
	if (frame.length > MAX_FRAME_CHARS) throw new Error('Peer transfer frame exceeds its size limit')
	while (channel.bufferedAmount > BUFFER_LIMIT_BYTES) {
	await new Promise<void>((resolve, reject) => {
	const timeout = window.setTimeout(() => reject(new Error('Peer data channel remained congested')), 15_000)
	channel.addEventListener('bufferedamountlow', () => { window.clearTimeout(timeout); resolve() }, { once: true })
	if (channel.readyState !== 'open') { window.clearTimeout(timeout); reject(new Error('Peer disconnected during transfer')) }
	})
	}
	if (channel.readyState !== 'open') throw new Error('Peer disconnected during transfer')
	channel.send(frame)
	}

	close() {
	this.closed = true
	for (const peer of this.peers.values()) peer.close()
	this.peers.clear(); this.channels.clear(); this.incoming.clear()
	this.signalQueues.clear()
	for (const pending of this.pending.values()) { window.clearTimeout(pending.timer); pending.reject(new Error('Node connection closed')) }
	this.pending.clear()
	for (const pending of this.pendingArtifactReads.values()) {
	window.clearTimeout(pending.timer)
	pending.reject(new Error('Node connection closed while reading a shard'))
	}
	this.pendingArtifactReads.clear()
	}
	}