test(control-plane): port workspace.test.ts to it.instance (#28572)

This commit is contained in:
Kit Langton
2026-05-21 03:23:26 +00:00
committed by GitHub
parent 8fc02b0130
commit d37f9e770b
3 changed files with 477 additions and 445 deletions
@@ -1,4 +1,5 @@
import { Context, Effect, FiberMap, Iterable, Layer, Schema, Stream } from "effect" import { Context, Effect, FiberMap, Iterable, Layer, Schema, Stream } from "effect"
import { serviceUse } from "@/effect/service-use"
import { FetchHttpClient, HttpBody, HttpClient, HttpClientError, HttpClientRequest } from "effect/unstable/http" import { FetchHttpClient, HttpBody, HttpClient, HttpClientError, HttpClientRequest } from "effect/unstable/http"
import { Database } from "@/storage/db" import { Database } from "@/storage/db"
import { asc } from "drizzle-orm" import { asc } from "drizzle-orm"
@@ -167,6 +168,8 @@ export interface Interface {
export class Service extends Context.Service<Service, Interface>()("@opencode/Workspace") {} export class Service extends Context.Service<Service, Interface>()("@opencode/Workspace") {}
export const use = serviceUse(Service)
export const layer = Layer.effect( export const layer = Layer.effect(
Service, Service,
Effect.gen(function* () { Effect.gen(function* () {
+3
View File
@@ -1,4 +1,5 @@
import { Slug } from "@opencode-ai/core/util/slug" import { Slug } from "@opencode-ai/core/util/slug"
import { serviceUse } from "@/effect/service-use"
import path from "path" import path from "path"
import { BackgroundJob } from "@/background/job" import { BackgroundJob } from "@/background/job"
import { BusEvent } from "@/bus/bus-event" import { BusEvent } from "@/bus/bus-event"
@@ -500,6 +501,8 @@ export interface Interface {
export class Service extends Context.Service<Service, Interface>()("@opencode/Session") {} export class Service extends Context.Service<Service, Interface>()("@opencode/Session") {}
export const use = serviceUse(Service)
export type Patch = Types.DeepMutable<SyncEvent.Event<typeof Event.Updated>["data"]["info"]> export type Patch = Types.DeepMutable<SyncEvent.Event<typeof Event.Updated>["data"]["info"]>
const db = <T>(fn: (d: Parameters<typeof Database.use>[0] extends (trx: infer D) => any ? D : never) => T) => const db = <T>(fn: (d: Parameters<typeof Database.use>[0] extends (trx: infer D) => any ? D : never) => T) =>
@@ -5,31 +5,31 @@ import Http from "node:http"
import path from "node:path" import path from "node:path"
import { setTimeout as delay } from "node:timers/promises" import { setTimeout as delay } from "node:timers/promises"
import { NodeHttpServer } from "@effect/platform-node" import { NodeHttpServer } from "@effect/platform-node"
import { Effect, Layer, Schema } from "effect" import { Effect, Exit, Fiber, Layer, Schema } from "effect"
import { FetchHttpClient, HttpServer, HttpServerRequest, HttpServerResponse } from "effect/unstable/http" import { FetchHttpClient, HttpServer, HttpServerRequest, HttpServerResponse } from "effect/unstable/http"
import { eq } from "drizzle-orm" import { eq } from "drizzle-orm"
import { AppFileSystem } from "@opencode-ai/core/filesystem" import { AppFileSystem } from "@opencode-ai/core/filesystem"
import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
import * as Log from "@opencode-ai/core/util/log" import * as Log from "@opencode-ai/core/util/log"
import { GlobalBus, type GlobalEvent } from "@/bus/global" import { GlobalBus, type GlobalEvent } from "@/bus/global"
import { Database } from "@/storage/db" import { Database } from "@/storage/db"
import { ProjectID } from "@/project/schema" import { ProjectID } from "@/project/schema"
import { ProjectTable } from "@/project/project.sql" import { ProjectTable } from "@/project/project.sql"
import { context, type InstanceContext } from "@/project/instance-context"
import { InstanceRef } from "@/effect/instance-ref" import { InstanceRef } from "@/effect/instance-ref"
import { InstanceState } from "@/effect/instance-state"
import { Session as SessionNs } from "@/session/session" import { Session as SessionNs } from "@/session/session"
import { SessionID } from "@/session/schema" import { SessionID } from "@/session/schema"
import { SessionTable } from "@/session/session.sql" import { SessionTable } from "@/session/session.sql"
import { SyncEvent } from "@/sync" import { SyncEvent } from "@/sync"
import { EventSequenceTable } from "@/sync/event.sql" import { EventSequenceTable } from "@/sync/event.sql"
import { resetDatabase } from "../fixture/db" import { resetDatabase } from "../fixture/db"
import { disposeAllInstances, provideTmpdirInstance, TestInstance, tmpdir } from "../fixture/fixture" import { disposeAllInstances, provideTmpdirInstance, TestInstance, tmpdirScoped } from "../fixture/fixture"
import { testEffect } from "../lib/effect" import { testEffect } from "../lib/effect"
import { registerAdapter } from "../../src/control-plane/adapters" import { registerAdapter } from "../../src/control-plane/adapters"
import { WorkspaceID } from "../../src/control-plane/schema" import { WorkspaceID } from "../../src/control-plane/schema"
import { WorkspaceTable } from "../../src/control-plane/workspace.sql" import { WorkspaceTable } from "../../src/control-plane/workspace.sql"
import type { Target, WorkspaceAdapter, WorkspaceInfo } from "../../src/control-plane/types" import type { Target, WorkspaceAdapter, WorkspaceInfo } from "../../src/control-plane/types"
import * as Workspace from "../../src/control-plane/workspace" import * as Workspace from "../../src/control-plane/workspace"
import { AppRuntime } from "@/effect/app-runtime"
import { InstanceStore } from "@/project/instance-store" import { InstanceStore } from "@/project/instance-store"
import { InstanceBootstrap } from "@/project/bootstrap" import { InstanceBootstrap } from "@/project/bootstrap"
import { Auth } from "@/auth" import { Auth } from "@/auth"
@@ -66,6 +66,8 @@ const testServerLayer = Layer.mergeAll(
NodeHttpServer.layer(Http.createServer, { host: "127.0.0.1", port: 0 }), NodeHttpServer.layer(Http.createServer, { host: "127.0.0.1", port: 0 }),
workspaceLayer(true), workspaceLayer(true),
SessionNs.defaultLayer, SessionNs.defaultLayer,
InstanceStore.defaultLayer.pipe(Layer.provide(InstanceBootstrap.defaultLayer)),
CrossSpawnSpawner.defaultLayer,
) )
const it = testEffect(testServerLayer) const it = testEffect(testServerLayer)
@@ -121,12 +123,6 @@ afterEach(async () => {
await resetDatabase() await resetDatabase()
}) })
async function withInstance<T>(fn: (ctx: InstanceContext) => T | Promise<T>) {
await using tmp = await tmpdir({ git: true })
const ctx = await AppRuntime.runPromise(InstanceStore.Service.use((store) => store.load({ directory: tmp.path })))
return await context.provide(ctx, () => fn(ctx))
}
async function initGitRepo(dir: string) { async function initGitRepo(dir: string) {
await fs.mkdir(dir, { recursive: true }) await fs.mkdir(dir, { recursive: true })
await $`git init`.cwd(dir).quiet() await $`git init`.cwd(dir).quiet()
@@ -139,47 +135,6 @@ async function initGitRepo(dir: string) {
await $`git commit -m "base"`.cwd(dir).quiet() await $`git commit -m "base"`.cwd(dir).quiet()
} }
function currentInstance() {
try {
return context.use()
} catch {
return undefined
}
}
const runWorkspace = <A, E>(effect: Effect.Effect<A, E, Workspace.Service>) => {
const ctx = currentInstance()
return AppRuntime.runPromise(ctx ? effect.pipe(Effect.provideService(InstanceRef, ctx)) : effect)
}
const createWorkspace = (input: Workspace.CreateInput) =>
runWorkspace(Workspace.Service.use((workspace) => workspace.create(input)))
const warpWorkspaceSession = (input: Workspace.SessionWarpInput) =>
runWorkspace(Workspace.Service.use((workspace) => workspace.sessionWarp(input)))
const listWorkspaces = (project: Parameters<Workspace.Interface["list"]>[0]) =>
runWorkspace(Workspace.Service.use((workspace) => workspace.list(project)))
const syncListWorkspaces = (project: Parameters<Workspace.Interface["syncList"]>[0]) =>
runWorkspace(Workspace.Service.use((workspace) => workspace.syncList(project)))
const getWorkspace = (id: WorkspaceID) => runWorkspace(Workspace.Service.use((workspace) => workspace.get(id)))
const removeWorkspace = (id: WorkspaceID) => runWorkspace(Workspace.Service.use((workspace) => workspace.remove(id)))
const workspaceStatus = () => runWorkspace(Workspace.Service.use((workspace) => workspace.status()))
const isWorkspaceSyncing = (id: WorkspaceID) =>
runWorkspace(Workspace.Service.use((workspace) => workspace.isSyncing(id)))
const startWorkspaceSyncing = (projectID: ProjectID) => {
void runWorkspace(Workspace.Service.use((workspace) => workspace.startWorkspaceSyncing(projectID)))
}
const startWorkspaceSyncingWithFlag = (projectID: ProjectID, experimentalWorkspaces: boolean) =>
Effect.runPromise(
Workspace.Service.use((workspace) => workspace.startWorkspaceSyncing(projectID)).pipe(
Effect.provide(workspaceLayer(experimentalWorkspaces)),
),
)
const waitForWorkspaceSync = (
workspaceID: WorkspaceID,
state: Record<string, number>,
signal?: AbortSignal,
timeout?: number,
) => runWorkspace(Workspace.Service.use((workspace) => workspace.waitForSync(workspaceID, state, signal, timeout)))
function captureGlobalEvents() { function captureGlobalEvents() {
const events: GlobalEvent[] = [] const events: GlobalEvent[] = []
const handler = (event: GlobalEvent) => events.push(event) const handler = (event: GlobalEvent) => events.push(event)
@@ -421,14 +376,20 @@ describe("workspace schemas and exports", () => {
}) })
describe("workspace CRUD", () => { describe("workspace CRUD", () => {
test("get returns undefined for a missing workspace", async () => { it.instance(
await withInstance(async () => { "get returns undefined for a missing workspace",
expect(await getWorkspace(WorkspaceID.ascending("wrk_missing_get"))).toBeUndefined() () =>
}) Effect.gen(function* () {
}) expect(yield* Workspace.use.get(WorkspaceID.ascending("wrk_missing_get"))).toBeUndefined()
}),
{ git: true },
)
test("list maps database rows, filters by project, and sorts by id", async () => { it.instance(
await withInstance(async (instance) => { "list maps database rows, filters by project, and sorts by id",
() =>
Effect.gen(function* () {
const instance = yield* InstanceState.context
const otherProjectID = ProjectID.make("project-other") const otherProjectID = ProjectID.make("project-other")
insertProject(otherProjectID, "/tmp/other") insertProject(otherProjectID, "/tmp/other")
const a = workspaceInfo(instance.project.id, "manual", { const a = workspaceInfo(instance.project.id, "manual", {
@@ -448,12 +409,16 @@ describe("workspace CRUD", () => {
insertWorkspace(other) insertWorkspace(other)
insertWorkspace(a) insertWorkspace(a)
expect(await listWorkspaces(instance.project)).toEqual([a, b]) expect(yield* Workspace.use.list(instance.project)).toEqual([a, b])
}) }),
}) { git: true },
)
test("create configures, persists, creates, starts local sync, and passes environment", async () => { it.instance(
await withInstance(async (instance) => { "create configures, persists, creates, starts local sync, and passes environment",
() =>
Effect.gen(function* () {
const instance = yield* InstanceState.context
process.env.OPENCODE_AUTH_CONTENT = JSON.stringify({ test: { type: "api", key: "secret" } }) process.env.OPENCODE_AUTH_CONTENT = JSON.stringify({ test: { type: "api", key: "secret" } })
process.env.OTEL_EXPORTER_OTLP_HEADERS = "authorization=otel" process.env.OTEL_EXPORTER_OTLP_HEADERS = "authorization=otel"
process.env.OTEL_EXPORTER_OTLP_ENDPOINT = "https://otel.test" process.env.OTEL_EXPORTER_OTLP_ENDPOINT = "https://otel.test"
@@ -481,7 +446,7 @@ describe("workspace CRUD", () => {
}) })
registerAdapter(instance.project.id, type, recorded.adapter) registerAdapter(instance.project.id, type, recorded.adapter)
const info = await createWorkspace({ const info = yield* Workspace.use.create({
id: workspaceID, id: workspaceID,
type, type,
branch: null, branch: null,
@@ -499,8 +464,8 @@ describe("workspace CRUD", () => {
projectID: instance.project.id, projectID: instance.project.id,
timeUsed: info.timeUsed, timeUsed: info.timeUsed,
}) })
expect(await getWorkspace(workspaceID)).toEqual(info) expect(yield* Workspace.use.get(workspaceID)).toEqual(info)
expect(await listWorkspaces(instance.project)).toEqual([info]) expect(yield* Workspace.use.list(instance.project)).toEqual([info])
expect(recorded.calls.configure).toHaveLength(1) expect(recorded.calls.configure).toHaveLength(1)
expect(recorded.calls.configure[0]).toMatchObject({ id: workspaceID, type, directory: null }) expect(recorded.calls.configure[0]).toMatchObject({ id: workspaceID, type, directory: null })
expect(recorded.calls.create).toHaveLength(1) expect(recorded.calls.create).toHaveLength(1)
@@ -521,15 +486,23 @@ describe("workspace CRUD", () => {
expect(recorded.calls.create[0].env.OTEL_EXPORTER_OTLP_HEADERS).toBe("authorization=otel") expect(recorded.calls.create[0].env.OTEL_EXPORTER_OTLP_HEADERS).toBe("authorization=otel")
expect(recorded.calls.create[0].env.OTEL_EXPORTER_OTLP_ENDPOINT).toBe("https://otel.test") expect(recorded.calls.create[0].env.OTEL_EXPORTER_OTLP_ENDPOINT).toBe("https://otel.test")
expect(recorded.calls.create[0].env.OTEL_RESOURCE_ATTRIBUTES).toBe("service.name=opencode-test") expect(recorded.calls.create[0].env.OTEL_RESOURCE_ATTRIBUTES).toBe("service.name=opencode-test")
expect((await workspaceStatus()).find((item) => item.workspaceID === workspaceID)?.status).toBe("connected") expect((yield* Workspace.use.status()).find((item) => item.workspaceID === workspaceID)?.status).toBe(
"connected",
)
await removeWorkspace(workspaceID) yield* Workspace.use.remove(workspaceID)
expect((await workspaceStatus()).find((item) => item.workspaceID === workspaceID)?.status).toBeUndefined() expect(
}) (yield* Workspace.use.status()).find((item) => item.workspaceID === workspaceID)?.status,
}) ).toBeUndefined()
}),
{ git: true },
)
test("create propagates configure failures and does not insert a workspace", async () => { it.instance(
await withInstance(async (instance) => { "create propagates configure failures and does not insert a workspace",
() =>
Effect.gen(function* () {
const instance = yield* InstanceState.context
const type = unique("configure-failure") const type = unique("configure-failure")
registerAdapter( registerAdapter(
instance.project.id, instance.project.id,
@@ -544,15 +517,21 @@ describe("workspace CRUD", () => {
}).adapter, }).adapter,
) )
await expect( const exit = yield* Workspace.use
createWorkspace({ type, branch: null, projectID: instance.project.id, extra: null }), .create({ type, branch: null, projectID: instance.project.id, extra: null })
).rejects.toThrow("configure exploded") .pipe(Effect.exit)
expect(await listWorkspaces(instance.project)).toEqual([]) expect(Exit.isFailure(exit)).toBe(true)
}) if (Exit.isFailure(exit)) expect(String(exit.cause)).toContain("configure exploded")
}) expect(yield* Workspace.use.list(instance.project)).toEqual([])
}),
{ git: true },
)
test("create leaves the inserted row when adapter create fails", async () => { it.instance(
await withInstance(async (instance) => { "create leaves the inserted row when adapter create fails",
() =>
Effect.gen(function* () {
const instance = yield* InstanceState.context
const type = unique("create-failure") const type = unique("create-failure")
const recorded = recordedAdapter({ const recorded = recordedAdapter({
async create() { async create() {
@@ -564,35 +543,45 @@ describe("workspace CRUD", () => {
}) })
registerAdapter(instance.project.id, type, recorded.adapter) registerAdapter(instance.project.id, type, recorded.adapter)
await expect( const exit = yield* Workspace.use
createWorkspace({ type, branch: "branch", projectID: instance.project.id, extra: { x: 1 } }), .create({ type, branch: "branch", projectID: instance.project.id, extra: { x: 1 } })
).rejects.toThrow("create exploded") .pipe(Effect.exit)
expect(Exit.isFailure(exit)).toBe(true)
if (Exit.isFailure(exit)) expect(String(exit.cause)).toContain("create exploded")
const rows = await listWorkspaces(instance.project) const rows = yield* Workspace.use.list(instance.project)
expect(rows).toHaveLength(1) expect(rows).toHaveLength(1)
expect(rows[0]).toMatchObject({ type, branch: "branch", extra: { x: 1 } }) expect(rows[0]).toMatchObject({ type, branch: "branch", extra: { x: 1 } })
expect(recorded.calls.target).toHaveLength(0) expect(recorded.calls.target).toHaveLength(0)
await removeWorkspace(rows[0].id) yield* Workspace.use.remove(rows[0].id)
}) }),
}) { git: true },
)
test("create returns after a local workspace reports error", async () => { it.instance(
await withInstance(async (instance) => { "create returns after a local workspace reports error",
() =>
Effect.gen(function* () {
const instance = yield* InstanceState.context
const type = unique("local-error") const type = unique("local-error")
const missing = path.join(instance.directory, "missing-local-target") const missing = path.join(instance.directory, "missing-local-target")
const recorded = localAdapter(missing, { createDir: false }) const recorded = localAdapter(missing, { createDir: false })
registerAdapter(instance.project.id, type, recorded.adapter) registerAdapter(instance.project.id, type, recorded.adapter)
const info = await createWorkspace({ type, branch: null, projectID: instance.project.id, extra: null }) const info = yield* Workspace.use.create({ type, branch: null, projectID: instance.project.id, extra: null })
expect(info.directory).toBe(missing) expect(info.directory).toBe(missing)
expect((await workspaceStatus()).find((item) => item.workspaceID === info.id)?.status).toBe("error") expect((yield* Workspace.use.status()).find((item) => item.workspaceID === info.id)?.status).toBe("error")
await removeWorkspace(info.id) yield* Workspace.use.remove(info.id)
}) }),
}) { git: true },
)
test("syncList registers adapter-listed workspaces that are missing by name", async () => { it.instance(
await withInstance(async (instance) => { "syncList registers adapter-listed workspaces that are missing by name",
() =>
Effect.gen(function* () {
const instance = yield* InstanceState.context
const type = unique("list-sync") const type = unique("list-sync")
const existing = workspaceInfo(instance.project.id, type, { const existing = workspaceInfo(instance.project.id, type, {
id: WorkspaceID.ascending("wrk_list_sync_existing"), id: WorkspaceID.ascending("wrk_list_sync_existing"),
@@ -629,22 +618,26 @@ describe("workspace CRUD", () => {
}) })
registerAdapter(instance.project.id, type, recorded.adapter) registerAdapter(instance.project.id, type, recorded.adapter)
await syncListWorkspaces(instance.project) yield* Workspace.use.syncList(instance.project)
const synced = (await listWorkspaces(instance.project)).filter((item) => item.name === discovered.name) const synced = (yield* Workspace.use.list(instance.project)).filter((item) => item.name === discovered.name)
expect(synced).toHaveLength(1) expect(synced).toHaveLength(1)
expect(synced[0]).toMatchObject(discovered) expect(synced[0]).toMatchObject(discovered)
expect(synced[0]?.id).toStartWith("wrk_") expect(synced[0]?.id).toStartWith("wrk_")
expect(await listWorkspaces(instance.project)).toEqual(expect.arrayContaining([existing, synced[0]])) expect(yield* Workspace.use.list(instance.project)).toEqual(expect.arrayContaining([existing, synced[0]]))
expect(recorded.calls.list).toBe(1) expect(recorded.calls.list).toBe(1)
expect(recorded.calls.configure).toHaveLength(0) expect(recorded.calls.configure).toHaveLength(0)
expect(recorded.calls.create).toHaveLength(0) expect(recorded.calls.create).toHaveLength(0)
expect(recorded.calls.target).toHaveLength(1) expect(recorded.calls.target).toHaveLength(1)
}) }),
}) { git: true },
)
test("syncList calls every registered adapter with a list method", async () => { it.instance(
await withInstance(async (instance) => { "syncList calls every registered adapter with a list method",
() =>
Effect.gen(function* () {
const instance = yield* InstanceState.context
const typeA = unique("list-sync-a") const typeA = unique("list-sync-a")
const typeB = unique("list-sync-b") const typeB = unique("list-sync-b")
const adapterA = recordedAdapter({ const adapterA = recordedAdapter({
@@ -690,8 +683,8 @@ describe("workspace CRUD", () => {
registerAdapter(instance.project.id, typeB, adapterB.adapter) registerAdapter(instance.project.id, typeB, adapterB.adapter)
registerAdapter(instance.project.id, unique("list-sync-none"), noList.adapter) registerAdapter(instance.project.id, unique("list-sync-none"), noList.adapter)
await syncListWorkspaces(instance.project) yield* Workspace.use.syncList(instance.project)
const synced = await listWorkspaces(instance.project) const synced = yield* Workspace.use.list(instance.project)
expect( expect(
synced synced
@@ -702,8 +695,9 @@ describe("workspace CRUD", () => {
expect(adapterA.calls.list).toBe(1) expect(adapterA.calls.list).toBe(1)
expect(adapterB.calls.list).toBe(1) expect(adapterB.calls.list).toBe(1)
expect(noList.calls.list).toBe(0) expect(noList.calls.list).toBe(0)
}) }),
}) { git: true },
)
it.live("remote create connects to routed event and history endpoints", () => { it.live("remote create connects to routed event and history endpoints", () => {
const calls: FetchCall[] = [] const calls: FetchCall[] = []
@@ -755,11 +749,14 @@ describe("workspace CRUD", () => {
}) })
}) })
test("remove returns undefined for a missing workspace", async () => { it.instance(
await withInstance(async () => { "remove returns undefined for a missing workspace",
expect(await removeWorkspace(WorkspaceID.ascending("wrk_missing_remove"))).toBeUndefined() () =>
}) Effect.gen(function* () {
}) expect(yield* Workspace.use.remove(WorkspaceID.ascending("wrk_missing_remove"))).toBeUndefined()
}),
{ git: true },
)
it.instance( it.instance(
"remove deletes the workspace, associated sessions, adapter resources, and status", "remove deletes the workspace, associated sessions, adapter resources, and status",
@@ -795,8 +792,11 @@ describe("workspace CRUD", () => {
{ git: true }, { git: true },
) )
test("remove still deletes the row when the adapter cannot remove resources", async () => { it.instance(
await withInstance(async (instance) => { "remove still deletes the row when the adapter cannot remove resources",
() =>
Effect.gen(function* () {
const instance = yield* InstanceState.context
const type = unique("remove-throws") const type = unique("remove-throws")
const info = workspaceInfo(instance.project.id, type, { id: WorkspaceID.ascending("wrk_remove_throws") }) const info = workspaceInfo(instance.project.id, type, { id: WorkspaceID.ascending("wrk_remove_throws") })
registerAdapter( registerAdapter(
@@ -813,10 +813,11 @@ describe("workspace CRUD", () => {
) )
insertWorkspace(info) insertWorkspace(info)
expect(await removeWorkspace(info.id)).toEqual(info) expect(yield* Workspace.use.remove(info.id)).toEqual(info)
expect(await getWorkspace(info.id)).toBeUndefined() expect(yield* Workspace.use.get(info.id)).toBeUndefined()
}) }),
}) { git: true },
)
it.instance( it.instance(
"sessionWarp moves a session into a local workspace and claims ownership", "sessionWarp moves a session into a local workspace and claims ownership",
@@ -924,28 +925,25 @@ describe("workspace CRUD", () => {
{ git: true }, { git: true },
) )
test("sessionWarp detaches to the source project when invoked from a workspace instance", async () => { it.instance(
await withInstance(async (instance) => { "sessionWarp detaches to the source project when invoked from a workspace instance",
() =>
Effect.gen(function* () {
const instance = yield* InstanceState.context
const projectID = instance.project.id const projectID = instance.project.id
await using workspaceTmp = await tmpdir({ git: true }) const workspaceDir = yield* tmpdirScoped({ git: true })
const previousType = unique("warp-detach-workspace-instance") const previousType = unique("warp-detach-workspace-instance")
const previous = workspaceInfo(projectID, previousType) const previous = workspaceInfo(projectID, previousType)
insertWorkspace(previous) insertWorkspace(previous)
registerAdapter(projectID, previousType, localAdapter(workspaceTmp.path, { createDir: false }).adapter) registerAdapter(projectID, previousType, localAdapter(workspaceDir, { createDir: false }).adapter)
const session = await AppRuntime.runPromise( const session = yield* SessionNs.use.create({})
SessionNs.Service.use((svc) => svc.create({})).pipe(Effect.provideService(InstanceRef, instance)),
)
attachSessionToWorkspace(session.id, previous.id) attachSessionToWorkspace(session.id, previous.id)
const workspaceCtx = await AppRuntime.runPromise( const workspaceCtx = yield* InstanceStore.Service.use((store) => store.load({ directory: workspaceDir }))
InstanceStore.Service.use((store) => store.load({ directory: workspaceTmp.path })), expect(workspaceCtx.project.id).not.toBe(projectID)
) yield* Workspace.use
const workspaceProjectID = await context.provide(workspaceCtx, async () => { .sessionWarp({ workspaceID: null, sessionID: session.id })
const id = workspaceCtx.project.id .pipe(Effect.provideService(InstanceRef, workspaceCtx))
expect(id).not.toBe(projectID)
await warpWorkspaceSession({ workspaceID: null, sessionID: session.id })
return id
})
expect( expect(
Database.use((db) => Database.use((db) =>
@@ -957,9 +955,10 @@ describe("workspace CRUD", () => {
)?.workspaceID, )?.workspaceID,
).toBeNull() ).toBeNull()
expect(sessionSequenceOwner(session.id)).toBe(projectID) expect(sessionSequenceOwner(session.id)).toBe(projectID)
expect(sessionSequenceOwner(session.id)).not.toBe(workspaceProjectID) expect(sessionSequenceOwner(session.id)).not.toBe(workspaceCtx.project.id)
}) }),
}) { git: true },
)
it.live("sessionWarp syncs previous remote history, replays it, steals, and claims the sequence", () => { it.live("sessionWarp syncs previous remote history, replays it, steals, and claims the sequence", () => {
const calls: FetchCall[] = [] const calls: FetchCall[] = []
@@ -1072,7 +1071,12 @@ describe("workspace sync state", () => {
insertWorkspace(info) insertWorkspace(info)
registerAdapter(instance.project.id, type, localAdapter(path.join(dir, "flag-disabled")).adapter) registerAdapter(instance.project.id, type, localAdapter(path.join(dir, "flag-disabled")).adapter)
yield* Effect.promise(() => startWorkspaceSyncingWithFlag(instance.project.id, false)) // Isolated runtime with experimentalWorkspaces=false so we can verify the flag gates sync.
yield* Effect.promise(() =>
Effect.runPromise(
Workspace.use.startWorkspaceSyncing(instance.project.id).pipe(Effect.provide(workspaceLayer(false))),
),
)
yield* Effect.sleep("25 millis") yield* Effect.sleep("25 millis")
expect((yield* workspace.status()).find((item) => item.workspaceID === info.id)?.status).toBeUndefined() expect((yield* workspace.status()).find((item) => item.workspaceID === info.id)?.status).toBeUndefined()
@@ -1563,51 +1567,62 @@ describe("workspace sync state", () => {
}) })
describe("workspace waitForSync", () => { describe("workspace waitForSync", () => {
test("returns immediately for an empty fence", async () => { it.instance(
await withInstance(async () => { "returns immediately for an empty fence",
await expect(waitForWorkspaceSync(WorkspaceID.ascending("wrk_wait_empty"), {})).resolves.toBeUndefined() () =>
}) Effect.gen(function* () {
}) expect(yield* Workspace.use.waitForSync(WorkspaceID.ascending("wrk_wait_empty"), {})).toBeUndefined()
}),
{ git: true },
)
test("returns immediately when the stored sequence already satisfies the fence", async () => { it.instance(
await withInstance(async () => { "returns immediately when the stored sequence already satisfies the fence",
() =>
Effect.gen(function* () {
const sessionID = SessionID.descending("ses_wait_done") const sessionID = SessionID.descending("ses_wait_done")
Database.use((db) => db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 4 }).run()) Database.use((db) => db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 4 }).run())
await expect( expect(
waitForWorkspaceSync(WorkspaceID.ascending("wrk_wait_done"), { [sessionID]: 4 }), yield* Workspace.use.waitForSync(WorkspaceID.ascending("wrk_wait_done"), { [sessionID]: 4 }),
).resolves.toBeUndefined() ).toBeUndefined()
await expect( expect(
waitForWorkspaceSync(WorkspaceID.ascending("wrk_wait_done_2"), { [sessionID]: 3 }), yield* Workspace.use.waitForSync(WorkspaceID.ascending("wrk_wait_done_2"), { [sessionID]: 3 }),
).resolves.toBeUndefined() ).toBeUndefined()
}) }),
}) { git: true },
)
test("waits until the database reaches the requested sequence and a workspace event arrives", async () => { it.instance(
await withInstance(async () => { "waits until the database reaches the requested sequence and a workspace event arrives",
() =>
Effect.gen(function* () {
const workspaceID = WorkspaceID.ascending("wrk_wait_event") const workspaceID = WorkspaceID.ascending("wrk_wait_event")
const sessionID = SessionID.descending("ses_wait_event") const sessionID = SessionID.descending("ses_wait_event")
Database.use((db) => db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 1 }).run()) Database.use((db) => db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 1 }).run())
const waited = waitForWorkspaceSync(workspaceID, { [sessionID]: 2 }) const waited = yield* Workspace.use.waitForSync(workspaceID, { [sessionID]: 2 }).pipe(Effect.forkScoped)
await delay(10) yield* Effect.sleep("10 millis")
Database.use((db) => Database.use((db) =>
db.update(EventSequenceTable).set({ seq: 2 }).where(eq(EventSequenceTable.aggregate_id, sessionID)).run(), db.update(EventSequenceTable).set({ seq: 2 }).where(eq(EventSequenceTable.aggregate_id, sessionID)).run(),
) )
GlobalBus.emit("event", { workspace: workspaceID, payload: { type: "anything" } }) GlobalBus.emit("event", { workspace: workspaceID, payload: { type: "anything" } })
await expect(waited).resolves.toBeUndefined() expect(yield* Fiber.join(waited)).toBeUndefined()
}) }),
}) { git: true },
)
test("a sync event for a different workspace can also release the fence", async () => { it.instance(
await withInstance(async () => { "a sync event for a different workspace can also release the fence",
() =>
Effect.gen(function* () {
const workspaceID = WorkspaceID.ascending("wrk_wait_sync_any") const workspaceID = WorkspaceID.ascending("wrk_wait_sync_any")
const sessionID = SessionID.descending("ses_wait_sync_any") const sessionID = SessionID.descending("ses_wait_sync_any")
Database.use((db) => db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 0 }).run()) Database.use((db) => db.insert(EventSequenceTable).values({ aggregate_id: sessionID, seq: 0 }).run())
const waited = waitForWorkspaceSync(workspaceID, { [sessionID]: 1 }) const waited = yield* Workspace.use.waitForSync(workspaceID, { [sessionID]: 1 }).pipe(Effect.forkScoped)
await delay(10) yield* Effect.sleep("10 millis")
Database.use((db) => Database.use((db) =>
db.update(EventSequenceTable).set({ seq: 1 }).where(eq(EventSequenceTable.aggregate_id, sessionID)).run(), db.update(EventSequenceTable).set({ seq: 1 }).where(eq(EventSequenceTable.aggregate_id, sessionID)).run(),
) )
@@ -1616,36 +1631,47 @@ describe("workspace waitForSync", () => {
payload: { type: "sync" }, payload: { type: "sync" },
}) })
await expect(waited).resolves.toBeUndefined() expect(yield* Fiber.join(waited)).toBeUndefined()
}) }),
}) { git: true },
)
test("rejects with the abort reason when aborted", async () => { it.instance(
await withInstance(async () => { "rejects with the abort reason when aborted",
() =>
Effect.gen(function* () {
const abort = new AbortController() const abort = new AbortController()
const reason = new Error("caller aborted") const reason = new Error("caller aborted")
const waited = waitForWorkspaceSync( const waited = yield* Workspace.use
.waitForSync(
WorkspaceID.ascending("wrk_wait_abort"), WorkspaceID.ascending("wrk_wait_abort"),
{ [SessionID.descending("ses_wait_abort")]: 1 }, { [SessionID.descending("ses_wait_abort")]: 1 },
abort.signal, abort.signal,
) )
.pipe(Effect.flip, Effect.forkScoped)
abort.abort(reason) abort.abort(reason)
const error = yield* Fiber.join(waited)
await expect(waited).rejects.toMatchObject({ expect(error).toMatchObject({
_tag: "WorkspaceSyncAbortedError", _tag: "WorkspaceSyncAbortedError",
message: reason.message, message: reason.message,
cause: reason, cause: reason,
}) })
}) }),
}) { git: true },
)
test("times out with the requested fence in the error message", async () => { it.instance(
await withInstance(async () => { "times out with the requested fence in the error message",
() =>
Effect.gen(function* () {
const sessionID = SessionID.descending("ses_wait_timeout") const sessionID = SessionID.descending("ses_wait_timeout")
await expect( const failure = yield* Workspace.use
waitForWorkspaceSync(WorkspaceID.ascending("wrk_wait_timeout"), { [sessionID]: 1 }, undefined, 25), .waitForSync(WorkspaceID.ascending("wrk_wait_timeout"), { [sessionID]: 1 }, undefined, 25)
).rejects.toThrow(`Timed out waiting for sync fence: {"${sessionID}":1}`) .pipe(Effect.flip)
}) expect(String(failure)).toContain(`Timed out waiting for sync fence: {"${sessionID}":1}`)
}, 7000) }),
{ git: true, config: undefined },
7000,
)
}) })