instance-state.test.ts 12 KB

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