Unified · Split
| @@ -1,69 +0,0 @@ | |||
| 1 | import { mkdtempSync } from "node:fs"; | ||
| 2 | import { tmpdir } from "node:os"; | ||
| 3 | import { join } from "node:path"; | ||
| 4 | |||
| 5 | import { describe, expect, it } from "@effect/vitest"; | ||
| 6 | import { Effect } from "effect"; | ||
| 7 | |||
| 8 | import { makeGitArtifactStore } from "./git-artifact-store"; | ||
| 9 | import { scopeAddress } from "../seams/scope-address"; | ||
| 10 | import type { ArtifactStoreError } from "../seams/artifact-store"; | ||
| 11 | |||
| 12 | const run = <A, E>(effect: Effect.Effect<A, E>): Promise<A> => Effect.runPromise(effect); | ||
| 13 | |||
| 14 | // --------------------------------------------------------------------------- | ||
| 15 | // Finding 6 regression (git compare-and-swap): two commits built from the SAME | ||
| 16 | // parent cannot both advance the branch. The ref update is a CAS against the | ||
| 17 | // expected old value, so the second raced writer gets a typed conflict instead | ||
| 18 | // of silently clobbering the first. This is deterministic (no timing race): we | ||
| 19 | // force both to share a parent by committing them against a store whose head we | ||
| 20 | // pin. | ||
| 21 | // --------------------------------------------------------------------------- | ||
| 22 | |||
| 23 | describe("git ArtifactStore ref CAS (Fix 6)", () => { | ||
| 24 | it("two concurrent commits from one parent: one wins, the other typed-conflicts", async () => { | ||
| 25 | const root = mkdtempSync(join(tmpdir(), "apps-cas-")); | ||
| 26 | const storeA = makeGitArtifactStore({ root }); | ||
| 27 | const scopeA = await run(storeA.forScope(scopeAddress("org", "s"))); | ||
| 28 | |||
| 29 | // Seed a parent so racers commit ON TOP of the same head. | ||
| 30 | await run(scopeA.commit(new Map([["tools/base.ts", "// base"]]), "base")); | ||
| 31 | |||
| 32 | // Many SEPARATE store instances over the SAME repo dir. The repo already | ||
| 33 | // exists (seeded above), so opening each scope is a re-`init --bare` (a no-op | ||
| 34 | // on an existing repo). Open them SEQUENTIALLY so the idempotent init doesn't | ||
| 35 | // itself race, then race only the COMMITS (the ref-CAS is what's under test). | ||
| 36 | const N = 12; | ||
| 37 | const openedScopes: { commit: typeof scopeA.commit; log: typeof scopeA.log }[] = []; | ||
| 38 | for (let i = 0; i < N; i++) { | ||
| 39 | const store = makeGitArtifactStore({ root }); | ||
| 40 | // eslint-disable-next-line no-await-in-loop | ||
| 41 | openedScopes.push(await run(store.forScope(scopeAddress("org", "s")))); | ||
| 42 | } | ||
| 43 | const scopeB = openedScopes[0]; | ||
| 44 | |||
| 45 | // Fire many concurrent commits, ALL built against the seeded base parent. | ||
| 46 | const results = await Promise.allSettled( | ||
| 47 | openedScopes.map((sc, i) => | ||
| 48 | run(sc.commit(new Map([[`tools/c${i}.ts`, `// ${i}`]]), `c${i}`)), | ||
| 49 | ), | ||
| 50 | ); | ||
| 51 | |||
| 52 | const fulfilled = results.filter((r) => r.status === "fulfilled").length; | ||
| 53 | const rejected = results.filter((r) => r.status === "rejected"); | ||
| 54 | |||
| 55 | // Any failure MUST be a typed conflict (never a silent clobber). | ||
| 56 | for (const r of rejected) { | ||
| 57 | const err = (r as PromiseRejectedResult).reason as ArtifactStoreError; | ||
| 58 | expect(String(JSON.stringify(err))).toMatch(/conflict/i); | ||
| 59 | } | ||
| 60 | |||
| 61 | // No lost commits: the git log length equals base(1) + every commit that | ||
| 62 | // reported success. Without the CAS, a raced writer clobbers the ref and an | ||
| 63 | // earlier successful commit vanishes from history, making the log SHORTER | ||
| 64 | // than base + fulfilled. The CAS guarantees every reported success is a real, | ||
| 65 | // reachable, non-clobbering advance. | ||
| 66 | const logEntries = await run(scopeB.log(1000)); | ||
| 67 | expect(logEntries.length).toBe(1 + fulfilled); | ||
| 68 | }, 60_000); | ||
| 69 | }); | ||
| @@ -2,9 +2,34 @@ import { mkdtempSync } from "node:fs"; · 1 unmodified line | |||
| 2 | import { tmpdir } from "node:os"; | 2 | import { tmpdir } from "node:os"; |
| 3 | import { join } from "node:path"; | 3 | import { join } from "node:path"; |
| 4 | 4 | ||
| 5 | import { describe, expect, it } from "@effect/vitest"; | ||
| 6 | import { Effect, Exit } from "effect"; | ||
| 7 | |||
| 5 | import { artifactStoreConformance } from "../seams/artifact-store.conformance"; | 8 | import { artifactStoreConformance } from "../seams/artifact-store.conformance"; |
| 6 | import { makeGitArtifactStore } from "./git-artifact-store"; | 9 | import { scopeAddress, scopeAddressStorageKey } from "../seams/scope-address"; |
| 10 | import { compareAndSwapSnapshotRef, makeGitArtifactStore } from "./git-artifact-store"; | ||
| 7 | 11 | ||
| 8 | artifactStoreConformance("git", () => | 12 | artifactStoreConformance("git", () => |
| 9 | makeGitArtifactStore({ root: mkdtempSync(join(tmpdir(), "apps-artifacts-")) }), | 13 | makeGitArtifactStore({ root: mkdtempSync(join(tmpdir(), "apps-artifacts-")) }), |
| 10 | ); | 14 | ); |
| 15 | |||
| 16 | const run = <A, E>(effect: Effect.Effect<A, E>): Promise<A> => Effect.runPromise(effect); | ||
| 17 | |||
| 18 | describe("git ArtifactStore CAS", () => { | ||
| 19 | it("returns a typed conflict when the expected head is stale", async () => { | ||
| 20 | const root = mkdtempSync(join(tmpdir(), "apps-artifacts-cas-")); | ||
| 21 | const address = scopeAddress("org", "s"); | ||
| 22 | const scope = await run(makeGitArtifactStore({ root }).forScope(address)); | ||
| 23 | const first = await run(scope.commit(new Map([["tools/a.ts", "a"]]), "first")); | ||
| 24 | await run(scope.commit(new Map([["tools/a.ts", "b"]]), "second")); | ||
| 25 | |||
| 26 | const repoDir = join(root, `${scopeAddressStorageKey(address)}.git`); | ||
| 27 | const exit = await Effect.runPromiseExit( | ||
| 28 | compareAndSwapSnapshotRef(repoDir, String(first.id), String(first.id)), | ||
| 29 | ); | ||
| 30 | |||
| 31 | expect(Exit.isFailure(exit)).toBe(true); | ||
| 32 | expect(JSON.stringify(exit)).toContain("publish conflict"); | ||
| 33 | expect(JSON.stringify(exit)).toContain('"conflict":true'); | ||
| 34 | }); | ||
| 35 | }); | ||
| @@ -69,6 +69,25 @@ const run = ( · 68 unmodified lines | |||
| 69 | } | 69 | } |
| 70 | }); | 70 | }); |
| 71 | 71 | ||
| 72 | export const compareAndSwapSnapshotRef = ( | ||
| 73 | repoDir: string, | ||
| 74 | commitHash: string, | ||
| 75 | expectedOld: string | null, | ||
| 76 | ): Effect.Effect<void, ArtifactStoreError> => | ||
| 77 | run(repoDir, ["update-ref", BRANCH, commitHash, expectedOld ?? EMPTY_OID]).pipe( | ||
| 78 | Effect.asVoid, | ||
| 79 | Effect.mapError( | ||
| 80 | (cause) => | ||
| 81 | new ArtifactStoreError({ | ||
| 82 | message: `publish conflict: ${BRANCH} moved concurrently (expected ${ | ||
| 83 | expectedOld ?? EMPTY_OID | ||
| 84 | }); retry from the new head`, | ||
| 85 | conflict: true, | ||
| 86 | cause, | ||
| 87 | }), | ||
| 88 | ), | ||
| 89 | ); | ||
| 90 | |||
| 72 | const makeScopeStore = (repoDir: string): ScopeArtifactStore => { | 91 | const makeScopeStore = (repoDir: string): ScopeArtifactStore => { |
| 73 | // Environment forcing a deterministic author/committer so a snapshot's hash | 92 | // Environment forcing a deterministic author/committer so a snapshot's hash |
| 74 | // depends only on its content + parent, not on wall-clock/identity drift. | 93 | // depends only on its content + parent, not on wall-clock/identity drift. |
| @@ -184,17 +203,7 @@ const makeScopeStore = (repoDir: string): ScopeArtifactStore => { · 109 unmodified lines | |||
| 184 | // is no parent, so we assert the ref is absent (the empty-oid form). A CAS | 203 | // is no parent, so we assert the ref is absent (the empty-oid form). A CAS |
| 185 | // failure surfaces as a typed conflict for the caller to retry from a | 204 | // failure surfaces as a typed conflict for the caller to retry from a |
| 186 | // fresh parent. | 205 | // fresh parent. |
| 187 | const expectedOld = parent ?? EMPTY_OID; | 206 | yield* compareAndSwapSnapshotRef(repoDir, commitHash, parent); |
| 188 | yield* run(repoDir, ["update-ref", BRANCH, commitHash, expectedOld]).pipe( | ||
| 189 | Effect.mapError( | ||
| 190 | (cause) => | ||
| 191 | new ArtifactStoreError({ | ||
| 192 | message: `publish conflict: ${BRANCH} moved concurrently (expected ${expectedOld}); retry from the new head`, | ||
| 193 | conflict: true, | ||
| 194 | cause, | ||
| 195 | }), | ||
| 196 | ), | ||
| 197 | ); | ||
| 198 | return yield* readMeta(commitHash); | 207 | return yield* readMeta(commitHash); |
| 199 | }), | 208 | }), |
| 200 | 209 | ||