feat(core): exponential backoff of workspace reconnect (#23083)
This commit is contained in:
@@ -229,6 +229,10 @@ export function DialogWorkspaceCreate(props: { onSelect: (workspaceID: string) =
|
|||||||
})
|
})
|
||||||
|
|
||||||
const result = await sdk.client.experimental.workspace.create({ type, branch: null }).catch((err) => {
|
const result = await sdk.client.experimental.workspace.create({ type, branch: null }).catch((err) => {
|
||||||
|
toast.show({
|
||||||
|
message: "Creating workspace failed",
|
||||||
|
variant: "error",
|
||||||
|
})
|
||||||
log.error("workspace create request failed", {
|
log.error("workspace create request failed", {
|
||||||
type,
|
type,
|
||||||
error: errorData(err),
|
error: errorData(err),
|
||||||
|
|||||||
@@ -34,7 +34,6 @@ export type Info = z.infer<typeof Info>
|
|||||||
export const ConnectionStatus = z.object({
|
export const ConnectionStatus = z.object({
|
||||||
workspaceID: WorkspaceID.zod,
|
workspaceID: WorkspaceID.zod,
|
||||||
status: z.enum(["connected", "connecting", "disconnected", "error"]),
|
status: z.enum(["connected", "connecting", "disconnected", "error"]),
|
||||||
error: z.string().optional(),
|
|
||||||
})
|
})
|
||||||
export type ConnectionStatus = z.infer<typeof ConnectionStatus>
|
export type ConnectionStatus = z.infer<typeof ConnectionStatus>
|
||||||
|
|
||||||
@@ -345,10 +344,10 @@ const connections = new Map<WorkspaceID, ConnectionStatus>()
|
|||||||
const aborts = new Map<WorkspaceID, AbortController>()
|
const aborts = new Map<WorkspaceID, AbortController>()
|
||||||
const TIMEOUT = 5000
|
const TIMEOUT = 5000
|
||||||
|
|
||||||
function setStatus(id: WorkspaceID, status: ConnectionStatus["status"], error?: string) {
|
function setStatus(id: WorkspaceID, status: ConnectionStatus["status"]) {
|
||||||
const prev = connections.get(id)
|
const prev = connections.get(id)
|
||||||
if (prev?.status === status && prev?.error === error) return
|
if (prev?.status === status) return
|
||||||
const next = { workspaceID: id, status, error }
|
const next = { workspaceID: id, status }
|
||||||
connections.set(id, next)
|
connections.set(id, next)
|
||||||
|
|
||||||
if (status === "error") {
|
if (status === "error") {
|
||||||
@@ -425,68 +424,78 @@ function route(url: string | URL, path: string) {
|
|||||||
return next
|
return next
|
||||||
}
|
}
|
||||||
|
|
||||||
async function syncWorkspace(space: Info, signal: AbortSignal) {
|
async function connectSSE(url: URL | string, headers: HeadersInit | undefined, signal: AbortSignal) {
|
||||||
|
const res = await fetch(route(url, "/global/event"), {
|
||||||
|
method: "GET",
|
||||||
|
headers,
|
||||||
|
signal,
|
||||||
|
})
|
||||||
|
|
||||||
|
if (!res.ok) throw new Error(`Workspace sync HTTP failure: ${res.status}`)
|
||||||
|
if (!res.body) throw new Error("No response body from global sync")
|
||||||
|
|
||||||
|
return res.body
|
||||||
|
}
|
||||||
|
|
||||||
|
async function syncWorkspaceLoop(space: Info, signal: AbortSignal) {
|
||||||
|
const adaptor = await getAdaptor(space.projectID, space.type)
|
||||||
|
const target = await adaptor.target(space)
|
||||||
|
|
||||||
|
if (target.type === "local") return null
|
||||||
|
|
||||||
|
let attempt = 0
|
||||||
|
|
||||||
while (!signal.aborted) {
|
while (!signal.aborted) {
|
||||||
log.info("connecting to global sync", { workspace: space.name })
|
log.info("connecting to global sync", { workspace: space.name })
|
||||||
setStatus(space.id, "connecting")
|
setStatus(space.id, "connecting")
|
||||||
|
|
||||||
const adaptor = await getAdaptor(space.projectID, space.type)
|
let stream
|
||||||
const target = await adaptor.target(space)
|
try {
|
||||||
|
stream = await connectSSE(target.url, target.headers, signal)
|
||||||
if (target.type === "local") return
|
} catch (err) {
|
||||||
|
setStatus(space.id, "error")
|
||||||
const res = await fetch(route(target.url, "/global/event"), {
|
|
||||||
method: "GET",
|
|
||||||
headers: target.headers,
|
|
||||||
signal,
|
|
||||||
}).catch((err: unknown) => {
|
|
||||||
setStatus(space.id, "error", err instanceof Error ? err.message : String(err))
|
|
||||||
|
|
||||||
log.info("failed to connect to global sync", {
|
log.info("failed to connect to global sync", {
|
||||||
workspace: space.name,
|
workspace: space.name,
|
||||||
error: err,
|
err,
|
||||||
})
|
})
|
||||||
return undefined
|
|
||||||
})
|
|
||||||
|
|
||||||
if (!res || !res.ok || !res.body) {
|
|
||||||
const error = !res ? "No response from global sync" : `Global sync HTTP ${res.status}`
|
|
||||||
log.info("failed to connect to global sync", { workspace: space.name, error })
|
|
||||||
setStatus(space.id, "error", error)
|
|
||||||
await sleep(1000)
|
|
||||||
continue
|
|
||||||
}
|
}
|
||||||
|
|
||||||
log.info("global sync connected", { workspace: space.name })
|
if (stream) {
|
||||||
setStatus(space.id, "connected")
|
attempt = 0
|
||||||
|
|
||||||
await parseSSE(res.body, signal, (evt: any) => {
|
log.info("global sync connected", { workspace: space.name })
|
||||||
try {
|
setStatus(space.id, "connected")
|
||||||
if (!("payload" in evt)) return
|
|
||||||
|
|
||||||
if (evt.payload.type === "sync") {
|
await parseSSE(stream, signal, (evt: any) => {
|
||||||
SyncEvent.replay(evt.payload.syncEvent as SyncEvent.SerializedEvent)
|
try {
|
||||||
|
if (!("payload" in evt)) return
|
||||||
|
|
||||||
|
if (evt.payload.type === "sync") {
|
||||||
|
SyncEvent.replay(evt.payload.syncEvent as SyncEvent.SerializedEvent)
|
||||||
|
}
|
||||||
|
|
||||||
|
GlobalBus.emit("event", {
|
||||||
|
directory: evt.directory,
|
||||||
|
project: evt.project,
|
||||||
|
workspace: space.id,
|
||||||
|
payload: evt.payload,
|
||||||
|
})
|
||||||
|
} catch (err) {
|
||||||
|
log.info("failed to replay global event", {
|
||||||
|
workspaceID: space.id,
|
||||||
|
error: err,
|
||||||
|
})
|
||||||
}
|
}
|
||||||
|
})
|
||||||
|
|
||||||
GlobalBus.emit("event", {
|
log.info("disconnected from global sync: " + space.id)
|
||||||
directory: evt.directory,
|
setStatus(space.id, "disconnected")
|
||||||
project: evt.project,
|
}
|
||||||
workspace: space.id,
|
|
||||||
payload: evt.payload,
|
|
||||||
})
|
|
||||||
} catch (err) {
|
|
||||||
log.info("failed to replay global event", {
|
|
||||||
workspaceID: space.id,
|
|
||||||
error: err,
|
|
||||||
})
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
log.info("disconnected from global sync: " + space.id)
|
// Back off reconnect attempts up to 2 minutes while the workspace
|
||||||
setStatus(space.id, "disconnected")
|
// stays unavailable.
|
||||||
|
await sleep(Math.min(120_000, 1_000 * 2 ** attempt))
|
||||||
// TODO: Implement exponential backoff
|
attempt += 1
|
||||||
await sleep(1000)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -498,7 +507,7 @@ async function startSync(space: Info) {
|
|||||||
|
|
||||||
if (target.type === "local") {
|
if (target.type === "local") {
|
||||||
void Filesystem.exists(target.directory).then((exists) => {
|
void Filesystem.exists(target.directory).then((exists) => {
|
||||||
setStatus(space.id, exists ? "connected" : "error", exists ? undefined : "directory does not exist")
|
setStatus(space.id, exists ? "connected" : "error")
|
||||||
})
|
})
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -510,10 +519,10 @@ async function startSync(space: Info) {
|
|||||||
const abort = new AbortController()
|
const abort = new AbortController()
|
||||||
aborts.set(space.id, abort)
|
aborts.set(space.id, abort)
|
||||||
|
|
||||||
void syncWorkspace(space, abort.signal).catch((error) => {
|
void syncWorkspaceLoop(space, abort.signal).catch((error) => {
|
||||||
aborts.delete(space.id)
|
aborts.delete(space.id)
|
||||||
|
|
||||||
setStatus(space.id, "error", String(error))
|
setStatus(space.id, "error")
|
||||||
log.warn("workspace listener failed", {
|
log.warn("workspace listener failed", {
|
||||||
workspaceID: space.id,
|
workspaceID: space.id,
|
||||||
error,
|
error,
|
||||||
|
|||||||
Reference in New Issue
Block a user