Skip to content

Commit efb947e

Browse files
d-csd-cs
committed
style: apply oxfmt
1 parent 420328d commit efb947e

3 files changed

Lines changed: 89 additions & 96 deletions

File tree

‎internal-packages/run-engine/src/engine/systems/waitpointSystem.ts‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -495,7 +495,6 @@ export class WaitpointSystem {
495495
blockingWaitpoints
496496
);
497497

498-
499498
// 3. Get the run (run-ops scalars) + resolve its environment via the control-plane resolver,
500499
// so the run-ops DB can split without a cross-provider join.
501500
const run = await this.$.runStore.findRun(

‎internal-packages/run-engine/src/engine/waitpointCoordinator/storeCoordinator.test.ts‎

Lines changed: 23 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -1871,31 +1871,32 @@ describe("readCompletionEnvelopes", () => {
18711871
}
18721872
});
18731873

1874-
redisTest("carries an offloaded value as a ref, not as an inline value", async ({
1875-
redisOptions,
1876-
}) => {
1877-
const store = coordinator(redisOptions);
1878-
try {
1879-
await store.createIfAbsent({ record: record("w_ref"), status: "PENDING" });
1880-
await store.complete({
1881-
waitpointId: "w_ref",
1882-
completion: completion({
1883-
outputType: "application/store",
1884-
output: { ref: "store-key-1" },
1885-
}),
1886-
});
1874+
redisTest(
1875+
"carries an offloaded value as a ref, not as an inline value",
1876+
async ({ redisOptions }) => {
1877+
const store = coordinator(redisOptions);
1878+
try {
1879+
await store.createIfAbsent({ record: record("w_ref"), status: "PENDING" });
1880+
await store.complete({
1881+
waitpointId: "w_ref",
1882+
completion: completion({
1883+
outputType: "application/store",
1884+
output: { ref: "store-key-1" },
1885+
}),
1886+
});
18871887

1888-
const [envelope] = await store.readCompletionEnvelopes({
1889-
runId: "run_env",
1890-
waitpointIds: ["w_ref"],
1891-
});
1888+
const [envelope] = await store.readCompletionEnvelopes({
1889+
runId: "run_env",
1890+
waitpointIds: ["w_ref"],
1891+
});
18921892

1893-
expect(envelope?.outputRef).toBe("store-key-1");
1894-
expect(envelope?.output).toBeUndefined();
1895-
} finally {
1896-
await store.quit();
1893+
expect(envelope?.outputRef).toBe("store-key-1");
1894+
expect(envelope?.output).toBeUndefined();
1895+
} finally {
1896+
await store.quit();
1897+
}
18971898
}
1898-
});
1899+
);
18991900

19001901
// The omission is the contract. A pending waitpoint has no envelope, and defaulting one
19011902
// here would hand the resolver a record it must not have. The caller's coverage check is

‎internal-packages/run-store/src/taskRunExecutionSnapshotStore.waitpointRecords.test.ts‎

Lines changed: 66 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -109,45 +109,42 @@ async function readRecords(
109109
}
110110

111111
describe("the completed-waitpoint record set", () => {
112-
containerTest("a mint writes the records the caller supplied", async ({
113-
prisma,
114-
redisOptions,
115-
}) => {
116-
const { decorated, redis } = build(prisma as never, redisOptions as never);
117-
const probe = createRedisClient(redisOptions, { onError: () => {} });
118-
try {
119-
const env = await seedSnapshotEnvironment(prisma);
120-
const runId = await seedRun(decorated, redis, env);
121-
const [wpA, wpB] = await seedSnapshotWaitpoints(prisma, env, 2);
122-
123-
await decorated.createExecutionSnapshot(
124-
resumeInput(
125-
runId,
126-
env,
127-
[
128-
{ id: wpA!, index: 0 },
129-
{ id: wpB!, index: 1 },
130-
],
131-
[record(wpA!), record(wpB!)]
132-
)
133-
);
134-
135-
const records = await readRecords(probe, runId);
136-
137-
expect(records).toHaveLength(2);
138-
expect(records?.map((r) => r.id).sort()).toEqual([wpA, wpB].sort());
139-
expect(records?.[0]?.output).toEqual({ inline: '{"ok":true}' });
140-
} finally {
141-
await Promise.all([redis.quit(), probe.quit().catch(() => {})]);
112+
containerTest(
113+
"a mint writes the records the caller supplied",
114+
async ({ prisma, redisOptions }) => {
115+
const { decorated, redis } = build(prisma as never, redisOptions as never);
116+
const probe = createRedisClient(redisOptions, { onError: () => {} });
117+
try {
118+
const env = await seedSnapshotEnvironment(prisma);
119+
const runId = await seedRun(decorated, redis, env);
120+
const [wpA, wpB] = await seedSnapshotWaitpoints(prisma, env, 2);
121+
122+
await decorated.createExecutionSnapshot(
123+
resumeInput(
124+
runId,
125+
env,
126+
[
127+
{ id: wpA!, index: 0 },
128+
{ id: wpB!, index: 1 },
129+
],
130+
[record(wpA!), record(wpB!)]
131+
)
132+
);
133+
134+
const records = await readRecords(probe, runId);
135+
136+
expect(records).toHaveLength(2);
137+
expect(records?.map((r) => r.id).sort()).toEqual([wpA, wpB].sort());
138+
expect(records?.[0]?.output).toEqual({ inline: '{"ok":true}' });
139+
} finally {
140+
await Promise.all([redis.quit(), probe.quit().catch(() => {})]);
141+
}
142142
}
143-
});
143+
);
144144

145145
// The inertness guarantee. A wait with no store-resident half supplies no records, and the
146146
// cycle key must then hold none — a Postgres-resident resume is unchanged.
147-
containerTest("a mint with no records supplied writes none", async ({
148-
prisma,
149-
redisOptions,
150-
}) => {
147+
containerTest("a mint with no records supplied writes none", async ({ prisma, redisOptions }) => {
151148
const { decorated, redis } = build(prisma as never, redisOptions as never);
152149
const probe = createRedisClient(redisOptions, { onError: () => {} });
153150
try {
@@ -174,12 +171,8 @@ describe("the completed-waitpoint record set", () => {
174171
const [wpA] = await seedSnapshotWaitpoints(prisma, env, 1);
175172

176173
const waitpoints = [{ id: wpA!, index: 0 }];
177-
await decorated.createExecutionSnapshot(
178-
resumeInput(runId, env, waitpoints, [record(wpA!)])
179-
);
180-
await decorated.createExecutionSnapshot(
181-
resumeInput(runId, env, waitpoints, [record(wpA!)])
182-
);
174+
await decorated.createExecutionSnapshot(resumeInput(runId, env, waitpoints, [record(wpA!)]));
175+
await decorated.createExecutionSnapshot(resumeInput(runId, env, waitpoints, [record(wpA!)]));
183176

184177
const cycleKeys = await probe.keys(`snap:{${runId}}:wp:*`);
185178

@@ -190,37 +183,37 @@ describe("the completed-waitpoint record set", () => {
190183
}
191184
});
192185

193-
containerTest("a record set survives beside a repeat-preserving order", async ({
194-
prisma,
195-
redisOptions,
196-
}) => {
197-
const { decorated, redis } = build(prisma as never, redisOptions as never);
198-
const probe = createRedisClient(redisOptions, { onError: () => {} });
199-
try {
200-
const env = await seedSnapshotEnvironment(prisma);
201-
const runId = await seedRun(decorated, redis, env);
202-
const [wpA] = await seedSnapshotWaitpoints(prisma, env, 1);
203-
204-
const created = await decorated.createExecutionSnapshot(
205-
resumeInput(
206-
runId,
207-
env,
208-
[
209-
{ id: wpA!, index: 0 },
210-
{ id: wpA!, index: 1 },
211-
],
212-
[record(wpA!)]
213-
)
214-
);
215-
216-
const ids = await redis.getSnapshotWaitpointIds(runId, created.id);
217-
218-
// One record, two positions. The record set carries membership, the order carries
219-
// multiplicity.
220-
expect(await readRecords(probe, runId)).toHaveLength(1);
221-
expect(ids.order).toEqual([wpA, wpA]);
222-
} finally {
223-
await Promise.all([redis.quit(), probe.quit().catch(() => {})]);
186+
containerTest(
187+
"a record set survives beside a repeat-preserving order",
188+
async ({ prisma, redisOptions }) => {
189+
const { decorated, redis } = build(prisma as never, redisOptions as never);
190+
const probe = createRedisClient(redisOptions, { onError: () => {} });
191+
try {
192+
const env = await seedSnapshotEnvironment(prisma);
193+
const runId = await seedRun(decorated, redis, env);
194+
const [wpA] = await seedSnapshotWaitpoints(prisma, env, 1);
195+
196+
const created = await decorated.createExecutionSnapshot(
197+
resumeInput(
198+
runId,
199+
env,
200+
[
201+
{ id: wpA!, index: 0 },
202+
{ id: wpA!, index: 1 },
203+
],
204+
[record(wpA!)]
205+
)
206+
);
207+
208+
const ids = await redis.getSnapshotWaitpointIds(runId, created.id);
209+
210+
// One record, two positions. The record set carries membership, the order carries
211+
// multiplicity.
212+
expect(await readRecords(probe, runId)).toHaveLength(1);
213+
expect(ids.order).toEqual([wpA, wpA]);
214+
} finally {
215+
await Promise.all([redis.quit(), probe.quit().catch(() => {})]);
216+
}
224217
}
225-
});
218+
);
226219
});

0 commit comments

Comments
 (0)