instance-state.test.ts 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394
  1. import { afterEach, expect } from "bun:test"
  2. import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
  3. import { $ } from "bun"
  4. import { Context, Deferred, Duration, Effect, Exit, Fiber, Layer } from "effect"
  5. import { InstanceState } from "@/effect/instance-state"
  6. import { InstanceStore } from "../../src/project/instance-store"
  7. import { Instance } from "../../src/project/instance"
  8. import { disposeAllInstances, provideInstance, tmpdirScoped } from "../fixture/fixture"
  9. import { testEffect } from "../lib/effect"
  10. const it = testEffect(CrossSpawnSpawner.defaultLayer)
  11. const access = <A, E>(state: InstanceState.InstanceState<A, E>, dir: string) =>
  12. InstanceState.get(state).pipe(provideInstance(dir))
  13. const tmpdirGitScoped = Effect.gen(function* () {
  14. const dir = yield* tmpdirScoped({ git: true })
  15. yield* Effect.promise(() => $`git commit --allow-empty --amend -m ${`root commit ${dir}`}`.cwd(dir).quiet())
  16. return dir
  17. })
  18. afterEach(async () => {
  19. await disposeAllInstances()
  20. })
  21. it.live("InstanceState caches values per directory", () =>
  22. Effect.gen(function* () {
  23. const dir = yield* tmpdirScoped()
  24. let n = 0
  25. const state = yield* InstanceState.make(() => Effect.sync(() => ({ n: ++n })))
  26. const a = yield* access(state, dir)
  27. const b = yield* access(state, dir)
  28. expect(a).toBe(b)
  29. expect(n).toBe(1)
  30. }),
  31. )
  32. it.live("InstanceState isolates directories", () =>
  33. Effect.gen(function* () {
  34. const one = yield* tmpdirScoped()
  35. const two = yield* tmpdirScoped()
  36. let n = 0
  37. const state = yield* InstanceState.make((dir) => Effect.sync(() => ({ dir, n: ++n })))
  38. const a = yield* access(state, one)
  39. const b = yield* access(state, two)
  40. const c = yield* access(state, one)
  41. expect(a).toBe(c)
  42. expect(a).not.toBe(b)
  43. expect(n).toBe(2)
  44. }),
  45. )
  46. it.live("InstanceState invalidates on reload", () =>
  47. Effect.gen(function* () {
  48. const dir = yield* tmpdirScoped()
  49. const seen: string[] = []
  50. let n = 0
  51. const state = yield* InstanceState.make(() =>
  52. Effect.acquireRelease(
  53. Effect.sync(() => ({ n: ++n })),
  54. (value) =>
  55. Effect.sync(() => {
  56. seen.push(String(value.n))
  57. }),
  58. ),
  59. )
  60. const a = yield* access(state, dir)
  61. yield* Effect.promise(() => InstanceStore.reloadInstance({ directory: dir }))
  62. const b = yield* access(state, dir)
  63. expect(a).not.toBe(b)
  64. expect(seen).toEqual(["1"])
  65. }),
  66. )
  67. it.live("InstanceState invalidates on disposeAll", () =>
  68. Effect.gen(function* () {
  69. const one = yield* tmpdirScoped()
  70. const two = yield* tmpdirScoped()
  71. const seen: string[] = []
  72. const state = yield* InstanceState.make((ctx) =>
  73. Effect.acquireRelease(
  74. Effect.sync(() => ({ dir: ctx.directory })),
  75. (value) =>
  76. Effect.sync(() => {
  77. seen.push(value.dir)
  78. }),
  79. ),
  80. )
  81. yield* access(state, one)
  82. yield* access(state, two)
  83. yield* Effect.promise(disposeAllInstances)
  84. expect(seen.sort()).toEqual([one, two].sort())
  85. }),
  86. )
  87. it.live("InstanceState.get reads the current directory lazily", () =>
  88. Effect.gen(function* () {
  89. const one = yield* tmpdirScoped()
  90. const two = yield* tmpdirScoped()
  91. interface Api {
  92. readonly get: () => Effect.Effect<string>
  93. }
  94. class Test extends Context.Service<Test, Api>()("@test/InstanceStateLazy") {
  95. static readonly layer = Layer.effect(
  96. Test,
  97. Effect.gen(function* () {
  98. const state = yield* InstanceState.make((ctx) => Effect.sync(() => ctx.directory))
  99. const get = InstanceState.get(state)
  100. return Test.of({
  101. get: Effect.fn("Test.get")(function* () {
  102. return yield* get
  103. }),
  104. })
  105. }),
  106. )
  107. }
  108. yield* Effect.gen(function* () {
  109. const a = yield* Test.use((svc) => svc.get()).pipe(provideInstance(one))
  110. const b = yield* Test.use((svc) => svc.get()).pipe(provideInstance(two))
  111. expect(a).toBe(one)
  112. expect(b).toBe(two)
  113. }).pipe(Effect.provide(Test.layer))
  114. }),
  115. )
  116. it.live("InstanceState preserves directory across async boundaries", () =>
  117. Effect.gen(function* () {
  118. const one = yield* tmpdirGitScoped
  119. const two = yield* tmpdirGitScoped
  120. const three = yield* tmpdirGitScoped
  121. interface Api {
  122. readonly get: () => Effect.Effect<{ directory: string; worktree: string; project: string }>
  123. }
  124. class Test extends Context.Service<Test, Api>()("@test/InstanceStateAsync") {
  125. static readonly layer = Layer.effect(
  126. Test,
  127. Effect.gen(function* () {
  128. const state = yield* InstanceState.make((ctx) =>
  129. Effect.sync(() => ({
  130. directory: ctx.directory,
  131. worktree: ctx.worktree,
  132. project: ctx.project.id,
  133. })),
  134. )
  135. return Test.of({
  136. get: Effect.fn("Test.get")(function* () {
  137. yield* Effect.promise(() => Bun.sleep(1))
  138. yield* Effect.sleep(Duration.millis(1))
  139. for (let i = 0; i < 100; i++) {
  140. yield* Effect.yieldNow
  141. }
  142. for (let i = 0; i < 100; i++) {
  143. yield* Effect.promise(() => Promise.resolve())
  144. }
  145. yield* Effect.sleep(Duration.millis(2))
  146. yield* Effect.promise(() => Bun.sleep(1))
  147. return yield* InstanceState.get(state)
  148. }),
  149. })
  150. }),
  151. )
  152. }
  153. yield* Effect.gen(function* () {
  154. const [a, b, c] = yield* Effect.all(
  155. [one, two, three].map((dir) => Test.use((svc) => svc.get()).pipe(provideInstance(dir))),
  156. { concurrency: "unbounded" },
  157. )
  158. expect(a).toEqual({ directory: one, worktree: one, project: a.project })
  159. expect(b).toEqual({ directory: two, worktree: two, project: b.project })
  160. expect(c).toEqual({ directory: three, worktree: three, project: c.project })
  161. expect(a.project).not.toBe(b.project)
  162. expect(a.project).not.toBe(c.project)
  163. expect(b.project).not.toBe(c.project)
  164. }).pipe(Effect.provide(Test.layer))
  165. }),
  166. )
  167. it.live("InstanceState survives high-contention concurrent access", () =>
  168. Effect.gen(function* () {
  169. const dirs = yield* Effect.all(
  170. Array.from({ length: 20 }, () => tmpdirScoped()),
  171. { concurrency: "unbounded" },
  172. )
  173. interface Api {
  174. readonly get: () => Effect.Effect<string>
  175. }
  176. class Test extends Context.Service<Test, Api>()("@test/HighContention") {
  177. static readonly layer = Layer.effect(
  178. Test,
  179. Effect.gen(function* () {
  180. const state = yield* InstanceState.make((ctx) => Effect.sync(() => ctx.directory))
  181. return Test.of({
  182. get: Effect.fn("Test.get")(function* () {
  183. for (let i = 0; i < 10; i++) {
  184. yield* Effect.promise(() => Bun.sleep(Math.random() * 3))
  185. yield* Effect.yieldNow
  186. yield* Effect.promise(() => Promise.resolve())
  187. }
  188. return yield* InstanceState.get(state)
  189. }),
  190. })
  191. }),
  192. )
  193. }
  194. yield* Effect.gen(function* () {
  195. const results = yield* Effect.all(
  196. dirs.map((dir) => Test.use((svc) => svc.get()).pipe(provideInstance(dir))),
  197. { concurrency: "unbounded" },
  198. )
  199. expect(results).toEqual(dirs)
  200. }).pipe(Effect.provide(Test.layer))
  201. }),
  202. )
  203. it.live("InstanceState correct after interleaved init and dispose", () =>
  204. Effect.gen(function* () {
  205. const one = yield* tmpdirScoped()
  206. const two = yield* tmpdirScoped()
  207. interface Api {
  208. readonly get: () => Effect.Effect<string>
  209. }
  210. class Test extends Context.Service<Test, Api>()("@test/InterleavedDispose") {
  211. static readonly layer = Layer.effect(
  212. Test,
  213. Effect.gen(function* () {
  214. const state = yield* InstanceState.make((ctx) =>
  215. Effect.promise(async () => {
  216. await Bun.sleep(5)
  217. return ctx.directory
  218. }),
  219. )
  220. return Test.of({
  221. get: Effect.fn("Test.get")(function* () {
  222. return yield* InstanceState.get(state)
  223. }),
  224. })
  225. }),
  226. )
  227. }
  228. yield* Effect.gen(function* () {
  229. const a = yield* Test.use((svc) => svc.get()).pipe(provideInstance(one))
  230. expect(a).toBe(one)
  231. const [, b] = yield* Effect.all(
  232. [
  233. Effect.promise(() => InstanceStore.reloadInstance({ directory: one })),
  234. Test.use((svc) => svc.get()).pipe(provideInstance(two)),
  235. ],
  236. { concurrency: "unbounded" },
  237. )
  238. expect(b).toBe(two)
  239. const c = yield* Test.use((svc) => svc.get()).pipe(provideInstance(one))
  240. expect(c).toBe(one)
  241. }).pipe(Effect.provide(Test.layer))
  242. }),
  243. )
  244. it.live("InstanceState mutation in one directory does not leak to another", () =>
  245. Effect.gen(function* () {
  246. const one = yield* tmpdirScoped()
  247. const two = yield* tmpdirScoped()
  248. const state = yield* InstanceState.make(() => Effect.sync(() => ({ count: 0 })))
  249. const s1 = yield* access(state, one)
  250. s1.count = 42
  251. const s2 = yield* access(state, two)
  252. expect(s2.count).toBe(0)
  253. const s1again = yield* access(state, one)
  254. expect(s1again.count).toBe(42)
  255. expect(s1again).toBe(s1)
  256. }),
  257. )
  258. it.live("InstanceState dedupes concurrent lookups", () =>
  259. Effect.gen(function* () {
  260. const dir = yield* tmpdirScoped()
  261. let n = 0
  262. const state = yield* InstanceState.make(() =>
  263. Effect.promise(async () => {
  264. n += 1
  265. await Bun.sleep(10)
  266. return { n }
  267. }),
  268. )
  269. const [a, b] = yield* Effect.all([access(state, dir), access(state, dir)], { concurrency: "unbounded" })
  270. expect(a).toBe(b)
  271. expect(n).toBe(1)
  272. }),
  273. )
  274. it.live("InstanceState survives deferred resume from the same instance context", () =>
  275. Effect.gen(function* () {
  276. const dir = yield* tmpdirScoped({ git: true })
  277. interface Api {
  278. readonly get: (gate: Deferred.Deferred<void>) => Effect.Effect<string>
  279. }
  280. class Test extends Context.Service<Test, Api>()("@test/DeferredResume") {
  281. static readonly layer = Layer.effect(
  282. Test,
  283. Effect.gen(function* () {
  284. const state = yield* InstanceState.make((ctx) => Effect.sync(() => ctx.directory))
  285. return Test.of({
  286. get: Effect.fn("Test.get")(function* (gate: Deferred.Deferred<void>) {
  287. yield* Deferred.await(gate)
  288. return yield* InstanceState.get(state)
  289. }),
  290. })
  291. }),
  292. )
  293. }
  294. yield* Effect.gen(function* () {
  295. const gate = yield* Deferred.make<void>()
  296. const fiber = yield* Test.use((svc) => svc.get(gate)).pipe(provideInstance(dir), Effect.forkScoped)
  297. yield* Deferred.succeed(gate, undefined).pipe(provideInstance(dir))
  298. const exit = yield* Fiber.await(fiber)
  299. expect(Exit.isSuccess(exit)).toBe(true)
  300. if (Exit.isSuccess(exit)) expect(exit.value).toBe(dir)
  301. }).pipe(Effect.provide(Test.layer))
  302. }),
  303. )
  304. it.live("InstanceState survives deferred resume outside ALS when InstanceRef is set", () =>
  305. Effect.gen(function* () {
  306. const dir = yield* tmpdirScoped({ git: true })
  307. interface Api {
  308. readonly get: (gate: Deferred.Deferred<void>) => Effect.Effect<string>
  309. }
  310. class Test extends Context.Service<Test, Api>()("@test/DeferredResumeOutside") {
  311. static readonly layer = Layer.effect(
  312. Test,
  313. Effect.gen(function* () {
  314. const state = yield* InstanceState.make((ctx) => Effect.sync(() => ctx.directory))
  315. return Test.of({
  316. get: Effect.fn("Test.get")(function* (gate: Deferred.Deferred<void>) {
  317. yield* Deferred.await(gate)
  318. return yield* InstanceState.get(state)
  319. }),
  320. })
  321. }),
  322. )
  323. }
  324. yield* Effect.gen(function* () {
  325. const gate = yield* Deferred.make<void>()
  326. const fiber = yield* Test.use((svc) => svc.get(gate)).pipe(provideInstance(dir), Effect.forkScoped)
  327. yield* Deferred.succeed(gate, undefined)
  328. const exit = yield* Fiber.await(fiber)
  329. expect(Exit.isSuccess(exit)).toBe(true)
  330. if (Exit.isSuccess(exit)) expect(exit.value).toBe(dir)
  331. }).pipe(Effect.provide(Test.layer))
  332. }),
  333. )