service.test.ts 8.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239
  1. import { NodeFileSystem } from "@effect/platform-node"
  2. import { afterEach, expect, test } from "bun:test"
  3. import { Effect } from "effect"
  4. import { mkdtemp, rm, writeFile } from "node:fs/promises"
  5. import { tmpdir } from "node:os"
  6. import { join } from "node:path"
  7. import { Service, type EnsureReason } from "../src/effect/service"
  8. const fixture = join(import.meta.dir, "fixture/service.ts")
  9. const processes: Bun.Subprocess[] = []
  10. const directories: string[] = []
  11. afterEach(async () => {
  12. processes.forEach((process) => process.kill("SIGTERM"))
  13. await Promise.all(processes.splice(0).map((process) => process.exited))
  14. await Promise.all(directories.splice(0).map((directory) => rm(directory, { recursive: true, force: true })))
  15. })
  16. test("a concurrent same-version start cannot invalidate a resolved endpoint", async () => {
  17. const directory = await temp()
  18. const registration = join(directory, "service.json")
  19. spawn(registration, "modern")
  20. await waitForFile(registration)
  21. const original = await Bun.file(registration).json()
  22. const starts: EnsureReason[] = []
  23. const first = run(
  24. Service.ensure({
  25. file: registration,
  26. version: "test",
  27. command: [],
  28. onStart: (reason) => starts.push(reason),
  29. }),
  30. )
  31. await waitForFile(registration + ".first-request")
  32. const resolved = await run(Service.ensure({ file: registration, version: "test" }))
  33. expect(resolved.url).toBe(original.url)
  34. await writeFile(registration + ".release", "")
  35. await first
  36. expect(starts).toEqual([])
  37. expect(await Bun.file(registration).json()).toEqual(original)
  38. expect(await health(resolved.url)).toEqual({ healthy: true, version: "test", pid: original.pid })
  39. })
  40. test("waits for a registered service to finish starting", async () => {
  41. const directory = await temp()
  42. const registration = join(directory, "service.json")
  43. const process = spawn(registration, "starting")
  44. await waitForFile(registration)
  45. const result = run(Service.ensure({ file: registration, version: "test", command: [] }))
  46. await Bun.sleep(500)
  47. expect(process.exitCode).toBe(null)
  48. await writeFile(registration + ".release", "")
  49. expect((await result).url).toBe((await Bun.file(registration).json()).url)
  50. })
  51. test("reports a failed registered service without spawning", async () => {
  52. const directory = await temp()
  53. const registration = join(directory, "service.json")
  54. const process = spawn(registration, "failed-owner")
  55. await waitForFile(registration)
  56. await expect(run(Service.ensure({ file: registration, version: "test", command: [] }))).rejects.toThrow(
  57. "Background service failed to start",
  58. )
  59. expect(process.exitCode).toBe(null)
  60. })
  61. test("requests graceful stop of the exact service instance", async () => {
  62. const directory = await temp()
  63. const registration = join(directory, "service.json")
  64. const process = spawn(registration, "graceful")
  65. await waitForFile(registration)
  66. const info = await Bun.file(registration).json()
  67. await run(Service.stop({ file: registration }))
  68. await process.exited
  69. expect(await Bun.file(registration + ".stop").json()).toEqual({ instanceID: info.id })
  70. })
  71. test("does not spawn contenders while an incompatible service rejects replacement", async () => {
  72. const directory = await temp()
  73. const registration = join(directory, "service.json")
  74. const contender = join(directory, "contender.json")
  75. const existing = spawn(registration, "reject-stop")
  76. await waitForFile(registration)
  77. const controller = new AbortController()
  78. const starting = Effect.runPromise(
  79. Service.ensure({
  80. file: registration,
  81. version: "test",
  82. command: [process.execPath, fixture, contender, "record-start"],
  83. }).pipe(Effect.provide(NodeFileSystem.layer)),
  84. { signal: controller.signal },
  85. )
  86. await waitForFile(registration + ".stop-attempt")
  87. await Bun.sleep(500)
  88. controller.abort()
  89. await starting.catch(() => undefined)
  90. expect(await Bun.file(contender + ".started").exists()).toBe(false)
  91. expect(existing.exitCode).toBe(null)
  92. })
  93. test("a legacy health response is still replaced", async () => {
  94. const directory = await temp()
  95. const registration = join(directory, "service.json")
  96. const existing = spawn(registration, "legacy")
  97. await waitForFile(registration)
  98. const starts: EnsureReason[] = []
  99. const result = run(Service.ensure({ file: registration, command: [], onStart: (reason) => starts.push(reason) }))
  100. await expect(result).rejects.toThrow("Missing service command")
  101. expect(starts).toEqual(["version-mismatch"])
  102. await existing.exited
  103. }, 10_000)
  104. test("waits for a slow winner while bounding lock probes", async () => {
  105. const directory = await temp()
  106. const registration = join(directory, "service.json")
  107. const endpoint = await run(
  108. Service.ensure({
  109. file: registration,
  110. version: "test",
  111. command: [process.execPath, fixture, registration, "coordinated"],
  112. }),
  113. )
  114. const info = await Bun.file(registration).json()
  115. try {
  116. expect(endpoint.url).toBe(info.url)
  117. expect(await health(endpoint.url)).toEqual({ healthy: true, version: "test", pid: info.pid })
  118. expect((await Bun.file(registration + ".starts").text()).trim().split("\n")).toHaveLength(2)
  119. } finally {
  120. process.kill(info.pid, "SIGTERM")
  121. }
  122. }, 15_000)
  123. test("reports a contender that fails to start", async () => {
  124. const directory = await temp()
  125. const registration = join(directory, "service.json")
  126. await expect(
  127. run(
  128. Service.ensure({
  129. file: registration,
  130. version: "test",
  131. command: [process.execPath, fixture, registration, "failed"],
  132. }),
  133. ),
  134. ).rejects.toThrow("Server process exited with code 1")
  135. }, 10_000)
  136. test("reports a contender terminated by a signal", async () => {
  137. const directory = await temp()
  138. const registration = join(directory, "service.json")
  139. await expect(
  140. run(
  141. Service.ensure({
  142. file: registration,
  143. version: "test",
  144. command: [process.execPath, fixture, registration, "signal"],
  145. }),
  146. ),
  147. ).rejects.toThrow(/Server process (terminated by|exited with code)/)
  148. }, 10_000)
  149. test("reports a slow contender that eventually fails", async () => {
  150. const directory = await temp()
  151. const registration = join(directory, "service.json")
  152. await expect(
  153. run(
  154. Service.ensure({
  155. file: registration,
  156. version: "test",
  157. command: [process.execPath, fixture, registration, "delayed-failed", "8000"],
  158. }),
  159. ),
  160. ).rejects.toThrow("Server process exited with code 1")
  161. }, 15_000)
  162. test("replaces an incompatible owner that appears during startup", async () => {
  163. const directory = await temp()
  164. const registration = join(directory, "service.json")
  165. const starting = run(
  166. Service.ensure({
  167. file: registration,
  168. version: "test",
  169. command: [process.execPath, fixture, registration, "delayed", "8000"],
  170. }),
  171. )
  172. await Bun.sleep(1_000)
  173. const old = spawn(registration, "old")
  174. await waitForFile(registration)
  175. const endpoint = await starting
  176. const info = await Bun.file(registration).json()
  177. try {
  178. expect(endpoint.url).toBe(info.url)
  179. expect(info.version).toBe("test")
  180. await old.exited
  181. } finally {
  182. process.kill(info.pid, "SIGTERM")
  183. }
  184. }, 20_000)
  185. function run<A, E>(effect: Effect.Effect<A, E>) {
  186. return Effect.runPromise(effect.pipe(Effect.provide(NodeFileSystem.layer)))
  187. }
  188. function spawn(registration: string, mode: string, ...args: string[]) {
  189. const subprocess = Bun.spawn([process.execPath, fixture, registration, mode, ...args], {
  190. stdout: "ignore",
  191. stderr: "inherit",
  192. })
  193. processes.push(subprocess)
  194. return subprocess
  195. }
  196. async function temp() {
  197. const directory = await mkdtemp(join(tmpdir(), "opencode-client-service-"))
  198. directories.push(directory)
  199. return directory
  200. }
  201. async function waitForFile(file: string) {
  202. for (let attempt = 0; attempt < 600; attempt++) {
  203. if (await Bun.file(file).exists()) return
  204. await Bun.sleep(5)
  205. }
  206. throw new Error(`Timed out waiting for ${file}`)
  207. }
  208. async function health(url: string) {
  209. return fetch(new URL("/api/health", url), { signal: AbortSignal.timeout(1_000) }).then((response) => response.json())
  210. }