Something went wrong. Try again.
source dump of claude code forked from oppi.li/claude-code
Something went wrong. Try again.
3.9 kB · 131 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132import { sleep } from '../../utils/sleep.js'
/** * Coalescing uploader for PUT /worker (session state + metadata). * * - 1 in-flight PUT + 1 pending patch * - New calls coalesce into pending (never grows beyond 1 slot) * - On success: send pending if exists * - On failure: exponential backoff (clamped), retries indefinitely * until success or close(). Absorbs any pending patches before each retry. * - No backpressure needed — naturally bounded at 2 slots * * Coalescing rules: * - Top-level keys (worker_status, external_metadata) — last value wins * - Inside external_metadata / internal_metadata — RFC 7396 merge: * keys are added/overwritten, null values preserved (server deletes) */
type WorkerStateUploaderConfig = { send: (body: Record<string, unknown>) => Promise<boolean> /** Base delay for exponential backoff (ms) */ baseDelayMs: number /** Max delay cap (ms) */ maxDelayMs: number /** Random jitter range added to retry delay (ms) */ jitterMs: number}
export class WorkerStateUploader { private inflight: Promise<void> | null = null private pending: Record<string, unknown> | null = null private closed = false private readonly config: WorkerStateUploaderConfig
constructor(config: WorkerStateUploaderConfig) { this.config = config }
/** * Enqueue a patch to PUT /worker. Coalesces with any existing pending * patch. Fire-and-forget — callers don't need to await. */ enqueue(patch: Record<string, unknown>): void { if (this.closed) return this.pending = this.pending ? coalescePatches(this.pending, patch) : patch void this.drain() }
close(): void { this.closed = true this.pending = null }
private async drain(): Promise<void> { if (this.inflight || this.closed) return if (!this.pending) return
const payload = this.pending this.pending = null
this.inflight = this.sendWithRetry(payload).then(() => { this.inflight = null if (this.pending && !this.closed) { void this.drain() } }) }
/** Retries indefinitely with exponential backoff until success or close(). */ private async sendWithRetry(payload: Record<string, unknown>): Promise<void> { let current = payload let failures = 0 while (!this.closed) { const ok = await this.config.send(current) if (ok) return
failures++ await sleep(this.retryDelay(failures))
// Absorb any patches that arrived during the retry if (this.pending && !this.closed) { current = coalescePatches(current, this.pending) this.pending = null } } }
private retryDelay(failures: number): number { const exponential = Math.min( this.config.baseDelayMs * 2 ** (failures - 1), this.config.maxDelayMs, ) const jitter = Math.random() * this.config.jitterMs return exponential + jitter }}
/** * Coalesce two patches for PUT /worker. * * Top-level keys: overlay replaces base (last value wins). * Metadata keys (external_metadata, internal_metadata): RFC 7396 merge * one level deep — overlay keys are added/overwritten, null values * preserved for server-side delete. */function coalescePatches( base: Record<string, unknown>, overlay: Record<string, unknown>,): Record<string, unknown> { const merged = { ...base }
for (const [key, value] of Object.entries(overlay)) { if ( (key === 'external_metadata' || key === 'internal_metadata') && merged[key] && typeof merged[key] === 'object' && typeof value === 'object' && value !== null ) { // RFC 7396 merge — overlay keys win, nulls preserved for server merged[key] = { ...(merged[key] as Record<string, unknown>), ...(value as Record<string, unknown>), } } else { merged[key] = value } }
return merged}