Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ import {
OrchestrationV2PlanArtifact,
OrchestrationV2TurnItemJson,
} from "@t3tools/contracts";
import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
import * as DateTime from "effect/DateTime";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
Expand Down Expand Up @@ -53,7 +54,12 @@ const make = Effect.gen(function* () {
const shell = (location: "active" | "archive" = "active", unsettledOnly = false) =>
wrap(
Effect.gen(function* () {
const snapshot = yield* projections.getShellSnapshot({ location, unsettledOnly });
// The full shell read takes seconds on a large database: run it on the
// SQLite reader worker, off the event loop and the writer. Every caller
// reads committed state; inside an open transaction it stays on the writer.
const snapshot = yield* projections
.getShellSnapshot({ location, unsettledOnly })
.pipe(NodeSqliteClient.readOnly);
// The V2 store returns archived rows in `archivedThreads`, never in `threads`.
const threads = location === "archive" ? snapshot.archivedThreads : snapshot.threads;
const names = new Map(
Expand Down
49 changes: 49 additions & 0 deletions apps/server/src/orchestration-v2/ThreadLaunchService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2308,6 +2308,55 @@ it.effect("cancels tracked setup before provider work is released", () =>
}),
);

it.effect("stops an in-flight branch rename when tracked setup is cancelled", () =>
Effect.gen(function* () {
const setupEntered = yield* Deferred.make<void>();
const branchNameStarted = yield* Deferred.make<void>();
const allowBranchName = yield* Deferred.make<void>();
const branchNameInterrupted = yield* Deferred.make<void>();
const renameAttempted = yield* Deferred.make<void>();
const harness = makeHarness({
renameBranch: (input) =>
Deferred.succeed(renameAttempted, undefined).pipe(Effect.as({ branch: input.newBranch })),
runSetup: () => Deferred.succeed(setupEntered, undefined).pipe(Effect.andThen(Effect.never)),
generateBranchName: () =>
Deferred.succeed(branchNameStarted, undefined).pipe(
Effect.andThen(Deferred.await(allowBranchName)),
Effect.as({ branch: "generated-branch" }),
Effect.onInterrupt(() => Deferred.succeed(branchNameInterrupted, undefined)),
),
});
yield* Effect.gen(function* () {
const launches = yield* ThreadLaunch.ThreadLaunchService;
const tracker = yield* WorktreeSetupTracker.WorktreeSetupTracker;
const threads = yield* ThreadManagement.ThreadManagementService;
const launched = yield* launches.launch(
launchInput({
command: "launch:cancel-renaming",
thread: "thread:cancel-renaming",
message: "Start",
workspace: { type: "worktree", baseRef: "main" },
}),
);
yield* Deferred.await(setupEntered);
yield* Deferred.await(branchNameStarted);
assert.isTrue(yield* tracker.cancel(launched.threadId));
// Let a rename that survived the cancel finish, then check it never ran.
yield* Deferred.succeed(allowBranchName, undefined);
yield* Effect.raceFirst(
Deferred.await(branchNameInterrupted),
Deferred.await(renameAttempted),
);
assert.isTrue(
yield* Deferred.isDone(branchNameInterrupted),
"the cancelled setup's branch rename kept running",
);
assert.equal(harness.renameBranch.mock.calls.length, 0);
assert.isNull((yield* threads.getThreadProjection(launched.threadId)).thread.worktreePath);
}).pipe(Effect.provide(harness.layer));
}),
);

it.effect.each([0, 1])("releases an async setup before its completion with exit %s", (exitCode) =>
Effect.gen(function* () {
const completion = yield* Deferred.make<{ exitCode: number | null; durationMs: number }>();
Expand Down
7 changes: 6 additions & 1 deletion apps/server/src/orchestration-v2/ThreadLaunchService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import * as Cause from "effect/Cause";
import * as Context from "effect/Context";
import * as Effect from "effect/Effect";
import * as Exit from "effect/Exit";
import * as Fiber from "effect/Fiber";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Ref from "effect/Ref";
Expand Down Expand Up @@ -259,6 +260,9 @@ const make = Effect.gen(function* () {
let createdWorktreePath: string | null = null;
let setupTerminalId: string | null = null;
let workspaceRecorded = false;
// The background branch rename records the worktree again when it lands,
// so cleanup stops it before forgetting a removed worktree.
let branchRename: Fiber.Fiber<unknown> | null = null;
if (input.workspaceStrategy.type === "worktree") {
yield* setupTracker.begin({
threadId,
Expand Down Expand Up @@ -446,7 +450,7 @@ const make = Effect.gen(function* () {
) {
const oldBranch = branch;
const worktreeCwd = worktreePath;
yield* generateBranchNameFor(worktreeCwd, initialMessage).pipe(
branchRename = yield* generateBranchNameFor(worktreeCwd, initialMessage).pipe(
Effect.flatMap(({ branch: newBranch, exactName }) =>
git.renameBranch({
cwd: worktreeCwd,
Expand Down Expand Up @@ -586,6 +590,7 @@ const make = Effect.gen(function* () {
yield* terminals
.close({ threadId, terminalId: setupTerminalId, deleteHistory: true })
.pipe(Effect.ignore);
if (branchRename !== null) yield* Fiber.interrupt(branchRename);
const removedPath = createdWorktreePath;
// The thread forgets the worktree only once it is gone; a failed
// removal leaves the directory for the user to clean up rather than
Expand Down
11 changes: 7 additions & 4 deletions apps/server/src/orchestration-v2/ThreadPullRequestService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import {
type ThreadLinkedPullRequest,
} from "@t3tools/contracts";
import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";
import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
import * as Cause from "effect/Cause";
// T3-CUSTOM(expbkt3): the idle-thread rotation reads the clock.
import * as Clock from "effect/Clock";
Expand Down Expand Up @@ -133,10 +134,12 @@ export const make = Effect.gen(function* () {
*/
const readThreadSnapshot = ({ threadId, backfill }: RefreshRequest) =>
threadId === null
? orchestrator.getShellSnapshot({
location: "active",
unsettledOnly: !(backfill || pendingBackfill.size > 0),
})
? orchestrator
.getShellSnapshot({
location: "active",
unsettledOnly: !(backfill || pendingBackfill.size > 0),
})
.pipe(NodeSqliteClient.readOnly)
: Effect.gen(function* () {
// Read the sequence first. The thread is then at least this new, so a
// sync guarded by the sequence is rejected rather than missing a change.
Expand Down
26 changes: 16 additions & 10 deletions apps/server/src/orchestration-v2/http.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@ import {
TurnItemId,
type OrchestrationProjectShell,
} from "@t3tools/contracts";
// T3-CUSTOM(expbkt3): shell snapshot reads run on the SQLite reader worker.
import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
import * as Effect from "effect/Effect";
// T3-CUSTOM(expbkt3): resolve the bound operator.
import * as Option from "effect/Option";
Expand Down Expand Up @@ -145,16 +147,20 @@ export const layer = HttpApiBuilder.group(
);

const loadShellSnapshot = Effect.fn("http.orchestration.loadShellSnapshot")(function* () {
const base = yield* sql.withTransaction(
Effect.gen(function* () {
const threads = yield* threadManagement.getShellSnapshot({ location: "active" });
return buildActiveShellSnapshot({
projects: yield* projectStore.listShells(),
threads,
snapshotSequence: yield* applicationEvents.latestApplicationSequence,
});
}),
);
// T3-CUSTOM(expbkt3): read on the SQLite reader worker, off the event loop and the
// writer (upstream does this in loadShellSnapshotParts, which stage does not have).
const base = yield* sql
.withTransaction(
Effect.gen(function* () {
const threads = yield* threadManagement.getShellSnapshot({ location: "active" });
return buildActiveShellSnapshot({
projects: yield* projectStore.listShells(),
threads,
snapshotSequence: yield* applicationEvents.latestApplicationSequence,
});
}),
)
.pipe(NodeSqliteClient.readOnly);
const projects = yield* enrichProjectShells(base.projects);
// T3-CUSTOM(expbkt3): HTTP shell visibility matches the native socket stream.
return filterNativeShell({ ...base, projects }, yield* currentActorUserId);
Expand Down
1 change: 1 addition & 0 deletions apps/server/src/persistence/Sqlite.ts
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ export const layerFromPath = Effect.fn("makeSqlitePersistenceLive")(function* (d
layerSetup,
NodeSqliteClient.layer({
filename: dbPath,
readerWorker: true,
spanAttributes: {
"db.name": path.basename(dbPath),
"service.name": "t3code-server",
Expand Down
28 changes: 18 additions & 10 deletions apps/server/src/ws.ts
Original file line number Diff line number Diff line change
Expand Up @@ -143,6 +143,8 @@ import {
FORK_WS_RPCS,
} from "@t3tools/contracts";
import { resolveServerBackgroundActivitySettings } from "@t3tools/shared/backgroundActivitySettings";
// T3-CUSTOM(expbkt3): shell reads run on the SQLite reader worker.
import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
import {
HttpRouter,
HttpServerRequest,
Expand Down Expand Up @@ -1015,16 +1017,20 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc
},
);
const loadSnapshot = Effect.fn("ws.orchestrationV2.loadShellSnapshot")(function* () {
const base = yield* sql.withTransaction(
Effect.gen(function* () {
const threads = yield* threadManagement.getShellSnapshot({ location: "active" });
return buildActiveShellSnapshot({
projects: yield* projects.listShells(),
threads,
snapshotSequence: yield* applicationEvents.latestApplicationSequence,
});
}),
);
// T3-CUSTOM(expbkt3): read on the SQLite reader worker, off the event loop and the
// writer (upstream does this in loadShellSnapshotParts, which stage does not have).
const base = yield* sql
.withTransaction(
Effect.gen(function* () {
const threads = yield* threadManagement.getShellSnapshot({ location: "active" });
return buildActiveShellSnapshot({
projects: yield* projects.listShells(),
threads,
snapshotSequence: yield* applicationEvents.latestApplicationSequence,
});
}),
)
.pipe(NodeSqliteClient.readOnly);
const enriched = yield* enrichProjectShells(base.projects);
return {
snapshot: { ...base, projects: enriched.projects } as OrchestrationV2ShellSnapshot,
Expand Down Expand Up @@ -1923,6 +1929,8 @@ const layerWsRpc = (
}),
)
.pipe(
// T3-CUSTOM(expbkt3): read on the SQLite reader worker; enrichment runs after it.
NodeSqliteClient.readOnly,
Effect.flatMap((snapshot) =>
enrichProjectShells(snapshot.projects).pipe(
Effect.map(({ projects }) => ({ ...snapshot, projects })),
Expand Down
Loading
Loading