service.test.ts 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289
  1. import { NodeFileSystem } from "@effect/platform-node"
  2. import { Service } from "@opencode-ai/client/effect/service"
  3. import { Database } from "@opencode-ai/core/database/database"
  4. import { EventV2 } from "@opencode-ai/core/event"
  5. import { EventTable } from "@opencode-ai/core/event/sql"
  6. import { Global } from "@opencode-ai/core/global"
  7. import { Project } from "@opencode-ai/core/project"
  8. import { ProjectTable } from "@opencode-ai/core/project/sql"
  9. import { AbsolutePath } from "@opencode-ai/core/schema"
  10. import { SessionV2 } from "@opencode-ai/core/session"
  11. import { SessionEvent } from "@opencode-ai/core/session/event"
  12. import { SessionTable } from "@opencode-ai/core/session/sql"
  13. import { expect, test } from "bun:test"
  14. import { Effect, Schedule, Schema } from "effect"
  15. import fs from "node:fs/promises"
  16. import os from "node:os"
  17. import path from "node:path"
  18. import { ServiceConfig } from "../src/services/service-config"
  19. test("local channel stores service config with the local service filename", async () => {
  20. const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-"))
  21. try {
  22. await Effect.runPromise(
  23. ServiceConfig.set("hostname", "127.0.0.2").pipe(
  24. Effect.provide(Global.layerWith({ config: path.join(root, "config"), state: path.join(root, "state") })),
  25. Effect.provide(NodeFileSystem.layer),
  26. ),
  27. )
  28. expect(await Bun.file(path.join(root, "config", "service-local.json")).json()).toEqual({
  29. hostname: "127.0.0.2",
  30. })
  31. expect(await Bun.file(path.join(root, "config", "service.json")).exists()).toBe(false)
  32. } finally {
  33. await fs.rm(root, { recursive: true, force: true })
  34. }
  35. })
  36. test("service filenames isolate installation channels", () => {
  37. expect(ServiceConfig.filename("latest")).toBe("service.json")
  38. expect(ServiceConfig.filename("local")).toBe("service-local.json")
  39. expect(ServiceConfig.filename("preview-a")).not.toBe(ServiceConfig.filename("preview-b"))
  40. expect(ServiceConfig.filename("preview-a")).not.toBe(ServiceConfig.filename("latest"))
  41. expect(ServiceConfig.versionBelongsToChannel("0.0.0-preview-a-1234", "preview-a")).toBe(true)
  42. expect(ServiceConfig.versionBelongsToChannel("0.0.0-preview-a-1234.2", "preview-a")).toBe(true)
  43. expect(ServiceConfig.versionBelongsToChannel("0.0.0-preview-a-other-1234", "preview-a")).toBe(false)
  44. expect(ServiceConfig.versionBelongsToChannel("1.2.3", "preview-a")).toBe(false)
  45. })
  46. test("preview registration migration never moves stable discovery", async () => {
  47. const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-migration-"))
  48. const legacy = path.join(root, "service.json")
  49. const target = path.join(root, ServiceConfig.filename("preview-a"))
  50. try {
  51. await fs.writeFile(
  52. legacy,
  53. JSON.stringify({ id: "old-preview", version: "0.0.0-preview-a-1234", url: "http://localhost:4096", pid: 1 }),
  54. )
  55. await Effect.runPromise(
  56. ServiceConfig.migrateRegistration(legacy, target, "preview-a", "0.0.0-preview-a-5678").pipe(
  57. Effect.provide(NodeFileSystem.layer),
  58. ),
  59. )
  60. expect(await Bun.file(legacy).exists()).toBe(true)
  61. expect(await Bun.file(target).json()).toMatchObject({ id: "old-preview" })
  62. await fs.rm(target)
  63. await fs.writeFile(legacy, JSON.stringify({ id: "stable", version: "1.2.3", url: "http://localhost:4096", pid: 1 }))
  64. await Effect.runPromise(
  65. ServiceConfig.migrateRegistration(legacy, target, "preview-a", "0.0.0-preview-a-5678").pipe(
  66. Effect.provide(NodeFileSystem.layer),
  67. ),
  68. )
  69. expect(await Bun.file(legacy).exists()).toBe(true)
  70. expect(await Bun.file(target).exists()).toBe(false)
  71. await fs.writeFile(
  72. legacy,
  73. JSON.stringify({ id: "old-preview", version: "0.0.0-preview-a-1234", url: "http://localhost:4096", pid: 1 }),
  74. )
  75. await fs.writeFile(target, JSON.stringify({ id: "current-preview" }))
  76. await Effect.runPromise(
  77. ServiceConfig.migrateRegistration(legacy, target, "preview-a", "0.0.0-preview-a-5678").pipe(
  78. Effect.provide(NodeFileSystem.layer),
  79. ),
  80. )
  81. expect(await Bun.file(legacy).exists()).toBe(true)
  82. expect(await Bun.file(target).json()).toMatchObject({ id: "current-preview" })
  83. } finally {
  84. await fs.rm(root, { recursive: true, force: true })
  85. }
  86. })
  87. test("concurrent service processes elect one server", async () => {
  88. const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-election-"))
  89. const database = path.join(root, "opencode.db")
  90. const env = {
  91. ...process.env,
  92. HOME: root,
  93. OPENCODE_DB: database,
  94. OPENCODE_TEST_HOME: root,
  95. XDG_CACHE_HOME: path.join(root, "cache"),
  96. XDG_CONFIG_HOME: path.join(root, "config"),
  97. XDG_DATA_HOME: path.join(root, "data"),
  98. XDG_STATE_HOME: path.join(root, "state"),
  99. }
  100. const sessionID = SessionV2.ID.make("ses_service_recovery")
  101. await withDatabase(
  102. database,
  103. Effect.gen(function* () {
  104. const { db } = yield* Database.Service
  105. yield* db
  106. .insert(ProjectTable)
  107. .values({ id: Project.ID.global, worktree: AbsolutePath.make(root), sandboxes: [] })
  108. .run()
  109. .pipe(Effect.orDie)
  110. yield* db
  111. .insert(SessionTable)
  112. .values({
  113. id: sessionID,
  114. project_id: Project.ID.global,
  115. slug: "recovery",
  116. directory: root,
  117. title: "recovery",
  118. version: "test",
  119. time_suspended: Date.now(),
  120. })
  121. .run()
  122. .pipe(Effect.orDie)
  123. }),
  124. )
  125. const command = [process.execPath, path.join(import.meta.dir, "../src/index.ts"), "serve", "--service"]
  126. const registration = path.join(root, "state", "opencode", "service-local.json")
  127. const processes = Array.from({ length: 10 }, () => Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" }))
  128. try {
  129. const info = await waitForInfo(registration)
  130. const winner = processes.find((process) => process.pid === info.pid)
  131. const losers = processes.filter((process) => process.pid !== info.pid)
  132. const exited = await Promise.all(
  133. losers.map((process) => Promise.race([process.exited.then(() => true), Bun.sleep(10_000).then(() => false)])),
  134. )
  135. expect(exited).toEqual(losers.map(() => true))
  136. expect(winner?.exitCode).toBe(null)
  137. expect(
  138. await fetch(new URL("/api/health", info.url), {
  139. headers: { authorization: "Basic " + btoa(`opencode:${info.password}`) },
  140. }).then((response) => response.json()),
  141. ).toEqual({
  142. healthy: true,
  143. version: info.version,
  144. pid: info.pid,
  145. })
  146. const blockedTemp = registration + "." + info.id + ".tmp"
  147. await fs.mkdir(blockedTemp)
  148. await fs.rm(registration)
  149. await Bun.sleep(6_000)
  150. expect(await Bun.file(registration).exists()).toBe(false)
  151. await fs.rm(blockedTemp, { recursive: true })
  152. const restored = await waitForInfo(registration)
  153. expect(restored.id).toBe(info.id)
  154. expect(restored.pid).toBe(info.pid)
  155. await fs.writeFile(registration, "not-json")
  156. const repaired = await waitForInfo(registration)
  157. expect(repaired.id).toBe(info.id)
  158. expect(repaired.pid).toBe(info.pid)
  159. const contender = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" })
  160. try {
  161. const contenderExited = await Promise.race([
  162. contender.exited.then(() => true),
  163. Bun.sleep(10_000).then(() => false),
  164. ])
  165. expect(contenderExited).toBe(true)
  166. expect((await waitForInfo(registration)).id).toBe(info.id)
  167. } finally {
  168. contender.kill("SIGTERM")
  169. await contender.exited
  170. }
  171. expect(
  172. await withDatabase(
  173. database,
  174. Effect.gen(function* () {
  175. const { db } = yield* Database.Service
  176. return yield* db
  177. .select({ timeSuspended: SessionTable.time_suspended })
  178. .from(SessionTable)
  179. .get()
  180. .pipe(Effect.orDie)
  181. }),
  182. ),
  183. ).toEqual({ timeSuspended: null })
  184. expect(await waitForExecutionStart(database, sessionID)).toBe(1)
  185. await Effect.runPromise(
  186. Service.stop({ file: registration }).pipe(Effect.provide(NodeFileSystem.layer)),
  187. )
  188. await winner?.exited
  189. } finally {
  190. processes.forEach((process) => process.kill("SIGTERM"))
  191. await Promise.all(processes.map((process) => process.exited))
  192. try {
  193. expect(await Bun.file(registration).exists()).toBe(false)
  194. } finally {
  195. await fs.rm(root, { recursive: true, force: true })
  196. }
  197. }
  198. }, 60_000)
  199. test("a failed service stays registered and owns the lock until stopped", async () => {
  200. const root = await fs.mkdtemp(path.join(os.tmpdir(), "opencode-service-failed-"))
  201. const database = path.join(root, "database")
  202. await fs.mkdir(database)
  203. const env = {
  204. ...process.env,
  205. HOME: root,
  206. OPENCODE_DB: database,
  207. OPENCODE_TEST_HOME: root,
  208. XDG_CACHE_HOME: path.join(root, "cache"),
  209. XDG_CONFIG_HOME: path.join(root, "config"),
  210. XDG_DATA_HOME: path.join(root, "data"),
  211. XDG_STATE_HOME: path.join(root, "state"),
  212. }
  213. const command = [process.execPath, path.join(import.meta.dir, "../src/index.ts"), "serve", "--service"]
  214. const registration = path.join(root, "state", "opencode", "service-local.json")
  215. const owner = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" })
  216. try {
  217. const info = await waitForInfo(registration)
  218. expect(owner.exitCode).toBe(null)
  219. const contender = Bun.spawn(command, { env, stderr: "pipe", stdout: "ignore" })
  220. expect(await Promise.race([contender.exited.then(() => true), Bun.sleep(10_000).then(() => false)])).toBe(true)
  221. expect((await waitForInfo(registration)).id).toBe(info.id)
  222. expect(owner.exitCode).toBe(null)
  223. await Effect.runPromise(Service.stop({ file: registration }).pipe(Effect.provide(NodeFileSystem.layer)))
  224. await owner.exited
  225. expect(await Bun.file(registration).exists()).toBe(false)
  226. } finally {
  227. owner.kill("SIGTERM")
  228. await owner.exited
  229. await fs.rm(root, { recursive: true, force: true })
  230. }
  231. }, 30_000)
  232. function withDatabase<A, E>(file: string, effect: Effect.Effect<A, E, Database.Service>) {
  233. return Effect.runPromise(effect.pipe(Effect.provide(Database.layerFromPath(file)), Effect.scoped))
  234. }
  235. function waitForExecutionStart(file: string, sessionID: SessionV2.ID) {
  236. return withDatabase(
  237. file,
  238. Effect.gen(function* () {
  239. const { db } = yield* Database.Service
  240. return yield* db
  241. .select({ id: EventTable.id, sessionID: EventTable.aggregate_id, type: EventTable.type })
  242. .from(EventTable)
  243. .all()
  244. .pipe(
  245. Effect.orDie,
  246. Effect.map((rows) =>
  247. rows.filter(
  248. (row) =>
  249. row.sessionID === sessionID &&
  250. row.type ===
  251. EventV2.versionedType(
  252. SessionEvent.Execution.Started.type,
  253. SessionEvent.Execution.Started.durable.version,
  254. ),
  255. ),
  256. ),
  257. Effect.filterOrFail((rows) => rows.length > 0),
  258. Effect.map((rows) => rows.length),
  259. Effect.retry(Schedule.spaced("50 millis").pipe(Schedule.both(Schedule.recurs(200)))),
  260. )
  261. }),
  262. )
  263. }
  264. async function waitForInfo(file: string) {
  265. for (let attempt = 0; attempt < 400; attempt++) {
  266. const value = await Bun.file(file)
  267. .json()
  268. .catch(() => undefined)
  269. if (value !== undefined) return Schema.decodeUnknownPromise(Service.Info)(value)
  270. await Bun.sleep(50)
  271. }
  272. throw new Error("Timed out waiting for service registration")
  273. }