Skip to content

perf(client): reduce thread-list update work #9716

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
t3dotgg merged 1 commit into from
Sep 4, 2026
+203 −23
Merged

perf(client): reduce thread-list update work #9716

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
169 changes: 161 additions & 8 deletions packages/client-runtime/src/state/entities.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ import { createEnvironmentSnapshotAtom } from "./snapshots.ts";
import { createEnvironmentThreadDetailAtoms } from "./threadDetail.ts";
import { mergeEnvironmentThread } from "./threadDetail.ts";
import { createEnvironmentThreadShellAtoms } from "./threadShell.ts";
import { applyShellStreamEvent } from "./shellReducer.ts";

const ENVIRONMENT_ID = EnvironmentId.make("environment-1");
const PROJECT_ID = ProjectId.make("project-1");
Expand Down Expand Up @@ -149,7 +150,7 @@ function shellState(snapshot: OrchestrationShellSnapshot): EnvironmentShellState
};
}

function makeHarness() {
function makeHarness(environmentIds: ReadonlyArray<EnvironmentId> = [ENVIRONMENT_ID]) {
const shellStateAtoms = Atom.family((_environmentId: EnvironmentId) =>
Atom.make(AsyncResult.success(shellState(SNAPSHOT))),
);
Expand All @@ -158,20 +159,20 @@ function makeHarness() {
);
const catalogValueAtom = Atom.make({
isReady: true,
entries: new Map([
[
ENVIRONMENT_ID,
entries: new Map(
environmentIds.map((environmentId) => [
environmentId,
{
target: new PrimaryConnectionTarget({
environmentId: ENVIRONMENT_ID,
environmentId,
label: "Environment",
httpBaseUrl: "https://example.test",
wsBaseUrl: "wss://example.test",
}),
profile: Option.none(),
},
],
]),
]),
),
});
const snapshotAtom = createEnvironmentSnapshotAtom(shellStateAtoms);
const projects = createEnvironmentProjectAtoms({
Expand All @@ -188,7 +189,9 @@ function makeHarness() {

return {
registry: AtomRegistry.make(),
catalogValueAtom,
shellStateAtom: shellStateAtoms(ENVIRONMENT_ID),
shellStateAtomForEnvironment: shellStateAtoms,
threadStateAtom: (threadId: ThreadId) => threadStateAtoms(`${ENVIRONMENT_ID}\u0000${threadId}`),
projects,
threadShells,
Expand Down Expand Up @@ -279,7 +282,8 @@ describe("environment entity projections", () => {
const refsByProjectAtom =
harness.threadShells.environmentThreadRefsByProjectAtom(ENVIRONMENT_ID);
const threadsAtom = harness.threadShells.threadShellsForProjectRefsAtom([projectRef]);
const refs = harness.registry.get(refsByProjectAtom).get(PROJECT_ID);
const membership = harness.registry.get(refsByProjectAtom);
const refs = membership.get(PROJECT_ID);
const threads = harness.registry.get(threadsAtom);

expect(threads).toHaveLength(1);
Expand All @@ -299,9 +303,158 @@ describe("environment entity projections", () => {
);

expect(harness.registry.get(refsByProjectAtom).get(PROJECT_ID)).toBe(refs);
expect(harness.registry.get(refsByProjectAtom)).toBe(membership);
expect(harness.registry.get(threadsAtom)).toBe(threads);
});

it("shares list values with point reads without retaining one atom per listed thread", () => {
const harness = makeHarness();
let snapshot: OrchestrationShellSnapshot = {
...SNAPSHOT,
threads: Array.from({ length: 200 }, (_, index) => ({
...THREAD_SHELL,
id: ThreadId.make(`listed-${index}`),
})),
};
harness.registry.set(harness.shellStateAtom, AsyncResult.success(shellState(snapshot)));
const listAtom = harness.threadShells.threadShellsAtom;
const projectListAtom = harness.threadShells.threadShellsForProjectRefsAtom([
{ environmentId: ENVIRONMENT_ID, projectId: PROJECT_ID },
]);
const disposeList = harness.registry.mount(listAtom);
const disposeProjectList = harness.registry.mount(projectListAtom);
try {
const before = harness.registry.get(listAtom);
expect(before).toHaveLength(200);
expect(harness.registry.get(projectListAtom)).toEqual(before);
expect(harness.registry.getNodes().size).toBeLessThan(20);
const firstAtom = harness.threadShells.threadShellAtom({
environmentId: ENVIRONMENT_ID,
threadId: snapshot.threads[0]!.id,
});
expect(harness.registry.get(firstAtom)).toBe(before[0]);

snapshot = applyShellStreamEvent(snapshot, {
kind: "thread-upserted",
sequence: 2,
thread: { ...snapshot.threads.at(-1)!, title: "Updated last thread" },
});
harness.registry.set(harness.shellStateAtom, AsyncResult.success(shellState(snapshot)));
const after = harness.registry.get(listAtom);
expect(after[0]).toBe(before[0]);
expect(after.at(-1)).not.toBe(before.at(-1));
expect(after.at(-1)?.title).toBe("Updated last thread");
expect(harness.registry.get(projectListAtom).at(-1)).toBe(after.at(-1));
expect(harness.registry.get(firstAtom)).toBe(after[0]);
expect(harness.registry.getNodes().size).toBeLessThan(20);
} finally {
disposeProjectList();
disposeList();
harness.registry.dispose();
}
});

it("keeps scoped identities and list order across project and environment changes", () => {
const remoteEnvironmentId = EnvironmentId.make("remote-environment");
const harness = makeHarness([ENVIRONMENT_ID, remoteEnvironmentId]);
const listAtom = harness.threadShells.threadShellsAtom;
const localRef = { environmentId: ENVIRONMENT_ID, threadId: THREAD_ID };
const remoteRef = { environmentId: remoteEnvironmentId, threadId: THREAD_ID };
const selectedAtom = harness.threadShells.threadShellsForProjectRefsAtom([
{ environmentId: remoteEnvironmentId, projectId: PROJECT_ID },
{ environmentId: ENVIRONMENT_ID, projectId: OTHER_PROJECT_ID },
{ environmentId: remoteEnvironmentId, projectId: PROJECT_ID },
{ environmentId: ENVIRONMENT_ID, projectId: PROJECT_ID },
]);
const disposeList = harness.registry.mount(listAtom);
const disposeSelected = harness.registry.mount(selectedAtom);
try {
const original = harness.registry.get(listAtom);
const local = harness.registry.get(harness.threadShells.threadShellAtom(localRef));
const remote = harness.registry.get(harness.threadShells.threadShellAtom(remoteRef));
expect(original).toHaveLength(4);
expect(original[0]).toBe(local);
expect(original[2]).toBe(remote);
expect(local).not.toBe(remote);
expect(local?.environmentId).toBe(ENVIRONMENT_ID);
expect(remote?.environmentId).toBe(remoteEnvironmentId);
expect(harness.registry.get(selectedAtom)).toEqual([remote, original[1], local]);

harness.registry.set(
harness.shellStateAtomForEnvironment(remoteEnvironmentId),
AsyncResult.success(shellState({ ...SNAPSHOT, threads: SNAPSHOT.threads.toReversed() })),
);
expect([
...harness.registry
.get(harness.threadShells.environmentThreadRefsByProjectAtom(remoteEnvironmentId))
.keys(),
]).toEqual([OTHER_PROJECT_ID, PROJECT_ID]);
harness.registry.set(
harness.shellStateAtomForEnvironment(remoteEnvironmentId),
AsyncResult.success(shellState(SNAPSHOT)),
);

let snapshot = applyShellStreamEvent(SNAPSHOT, {
kind: "thread-upserted",
sequence: 2,
thread: { ...THREAD_SHELL, projectId: OTHER_PROJECT_ID, title: "Moved thread" },
});
harness.registry.set(harness.shellStateAtom, AsyncResult.success(shellState(snapshot)));
const moved = harness.registry.get(harness.threadShells.threadShellAtom(localRef));
expect(harness.registry.get(selectedAtom)).toEqual([remote, moved, original[1]]);
expect(harness.registry.get(listAtom)[2]).toBe(remote);

snapshot = applyShellStreamEvent(snapshot, {
kind: "thread-removed",
sequence: 3,
threadId: OTHER_THREAD_ID,
});
harness.registry.set(harness.shellStateAtom, AsyncResult.success(shellState(snapshot)));
expect(harness.registry.get(selectedAtom)).toEqual([remote, moved]);
expect(
harness.registry.get(
harness.threadShells.threadShellAtom({
environmentId: ENVIRONMENT_ID,
threadId: OTHER_THREAD_ID,
}),
),
).toBeNull();

const createdId = ThreadId.make("created-thread");
snapshot = applyShellStreamEvent(snapshot, {
kind: "thread-upserted",
sequence: 4,
thread: { ...THREAD_SHELL, id: createdId },
});
harness.registry.set(harness.shellStateAtom, AsyncResult.success(shellState(snapshot)));
const created = harness.registry.get(listAtom)[1];
expect(created?.id).toBe(createdId);
expect(harness.registry.get(selectedAtom)).toEqual([remote, moved, created]);

const catalog = harness.registry.get(harness.catalogValueAtom);
harness.registry.set(harness.catalogValueAtom, {
...catalog,
entries: new Map([...catalog.entries].toReversed()),
});
expect(harness.registry.get(listAtom)).toEqual([remote, original[3], moved, created]);
harness.registry.set(harness.catalogValueAtom, {
...catalog,
entries: new Map([[remoteEnvironmentId, catalog.entries.get(remoteEnvironmentId)!]]),
});
expect(harness.registry.get(listAtom)).toEqual([remote, original[3]]);
harness.registry.set(
harness.shellStateAtomForEnvironment(remoteEnvironmentId),
AsyncResult.success(shellState({ ...SNAPSHOT, threads: [] })),
);
expect(harness.registry.get(listAtom)).toEqual([]);
expect(harness.registry.get(harness.threadShells.threadShellAtom(remoteRef))).toBeNull();
} finally {
disposeSelected();
disposeList();
harness.registry.dispose();
}
});

it("updates only the requested thread detail and preserves untouched field identities", () => {
const harness = makeHarness();
const threadRef = {
Expand Down
57 changes: 42 additions & 15 deletions packages/client-runtime/src/state/threadShell.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,26 @@ export function createEnvironmentThreadShellAtoms(input: {
environmentId: EnvironmentId,
) => Atom.Atom<OrchestrationShellSnapshot | null>;
}) {
// Point reads and aggregate lists share values without keeping an atom alive
// for every listed thread. Replaced source objects can be collected.
const scopedThreads = new WeakMap<
OrchestrationThreadShell,
Map<EnvironmentId, EnvironmentThreadShell>
>();
const scopedThread = (environmentId: EnvironmentId, thread: OrchestrationThreadShell) => {
let byEnvironment = scopedThreads.get(thread);
if (byEnvironment === undefined) {
byEnvironment = new Map();
scopedThreads.set(thread, byEnvironment);
}
let value = byEnvironment.get(environmentId);
if (value === undefined) {
value = scopeThreadShell(environmentId, thread);
byEnvironment.set(environmentId, value);
}
return value;
};

const environmentThreadsAtom = Atom.family((environmentId: EnvironmentId) =>
Atom.make(
(get): ReadonlyArray<OrchestrationThreadShell> =>
Expand Down Expand Up @@ -95,23 +115,26 @@ export function createEnvironmentThreadShellAtoms(input: {
previousRefs !== undefined && threadRefsEqual(previousRefs, refs) ? previousRefs : refs,
);
}
const previousProjectIds = [...previous.keys()];
if (
next.size === previous.size &&
[...next].every(
([projectId, refs], index) =>
previousProjectIds[index] === projectId && previous.get(projectId) === refs,
)
) {
return previous;
}
previous = next;
return previous;
}).pipe(Atom.withLabel(`environment-thread-refs-by-project:${environmentId}`));
});

const threadShellAtomFamily = Atom.family((key: string) => {
const ref = parseThreadKey(key);
let previousSource: OrchestrationThreadShell | null = null;
let previousValue: EnvironmentThreadShell | null = null;
return Atom.make((get) => {
const source = get(environmentThreadIndexAtom(ref.environmentId)).get(ref.threadId) ?? null;
if (source === previousSource) {
return previousValue;
}
previousSource = source;
previousValue = source === null ? null : scopeThreadShell(ref.environmentId, source);
return previousValue;
return source === null ? null : scopedThread(ref.environmentId, source);
}).pipe(Atom.withLabel(`environment-thread-shell:${key}`));
});

Expand All @@ -126,15 +149,17 @@ export function createEnvironmentThreadShellAtoms(input: {
get(environmentThreadRefsByProjectAtom(projectRef.environmentId)).get(
projectRef.projectId,
) ?? EMPTY_SCOPED_THREAD_REFS;
if (refs.length === 0) continue;
const threads = get(environmentThreadIndexAtom(projectRef.environmentId));
for (const ref of refs) {
const key = threadKey(ref);
if (seen.has(key)) {
continue;
}
seen.add(key);
const thread = get(threadShellAtomFamily(key));
if (thread !== null) {
next.push(thread);
const thread = threads.get(ref.threadId);
if (thread !== undefined) {
next.push(scopedThread(ref.environmentId, thread));
}
}
}
Expand All @@ -161,10 +186,12 @@ export function createEnvironmentThreadShellAtoms(input: {

let previousThreadShells: ReadonlyArray<EnvironmentThreadShell> = [];
const threadShellsAtom = Atom.make((get) => {
const next = get(threadRefsAtom).flatMap((ref) => {
const thread = get(threadShellAtomFamily(threadKey(ref)));
return thread === null ? [] : [thread];
});
const next: EnvironmentThreadShell[] = [];
for (const environmentId of get(input.catalogValueAtom).entries.keys()) {
for (const thread of get(environmentThreadsAtom(environmentId))) {
next.push(scopedThread(environmentId, thread));
}
}
if (arrayElementsEqual(previousThreadShells, next)) {
return previousThreadShells;
}
Expand Down
Loading