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
19 changes: 18 additions & 1 deletion packages/core/src/remote/replica.ts
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ import { logger } from "../log";
import { now } from "../util";
import { listActiveRuns } from "../runner/runner";
import { isSecondary, primaryRunnerId } from "./group";
import { LOCAL_SETTINGS, SETUP_TABLES, unsafe, withOwnFields } from "./snapshot";
import { freeSlug, LOCAL_SETTINGS, SETUP_TABLES, SLUG_SCOPES, slugHolder, unsafe, withOwnFields } from "./snapshot";

const log = logger("replica");

Expand Down Expand Up @@ -241,6 +241,18 @@ export function asPeer<T>(origin: string, at: string | null, fn: () => T): T {
}
}

/** Run `fn` as this computer's own change, also inside `asPeer`: what it writes goes to the other computers. */
export function asOwn<T>(fn: () => T): T {
const ctx = get<{ origin: string | null; at: string | null }>("SELECT origin, at FROM replica_ctx WHERE id = 1");
if (!ctx) return fn();
sql("UPDATE replica_ctx SET origin = NULL, at = NULL WHERE id = 1");
try {
return fn();
} finally {
sql("UPDATE replica_ctx SET origin = ?, at = ? WHERE id = 1", ctx.origin, ctx.at);
}
}

/** The last change this computer counted (local or taken from another one). */
export function replicaSeq(): number {
try {
Expand Down Expand Up @@ -574,6 +586,11 @@ function applyOne(peer: string, c: ReplicaChange, opts: ApplyOptions, incoming:
opts.onMissing?.(missing.table, missing.id);
return;
}
// Both computers made "earshot" while apart: the one that comes second here becomes "earshot-2" (the setup copy
// takes the new slug back to the peer). Only once the whole pull is in: a slug may be free by its end (a rename).
if (opts.retry && SLUG_SCOPES[t.table] && t.key.length === 1 && t.key[0] === "id" && row.slug != null && slugHolder(t.table, row, c.k)) {
row.slug = freeSlug(t.table, row, c.k);
}
const isNew = !get(`SELECT 1 FROM ${q(t.table)} WHERE ${keyWhere(t)}`, ...parts);
const extra = isNew ? (t.onInsert?.(row) ?? {}) : {};
const names = Object.keys(row);
Expand Down
15 changes: 10 additions & 5 deletions packages/core/src/remote/runners.ts
Original file line number Diff line number Diff line change
Expand Up @@ -747,9 +747,10 @@ function logFailure(text: string, err: unknown): void {

/**
* Bring a runner up to date: the setup when it differs (or `force`), then — for a chat that starts — the agent's memory
* and the browser sessions of the profile it uses. Returns what was done; a failed setup copy throws.
* and the browser sessions of the profile it uses. Returns what was done; a failed setup copy throws, except for `work`
* on a runner that has an earlier setup and the vault key.
*/
export function syncRunner(id: string, opts: { agentId?: string; profileId?: string | null; force?: boolean } = {}): Promise<{ config: boolean; memory: boolean; cookies: boolean; warnings: string[] }> {
export function syncRunner(id: string, opts: { agentId?: string; profileId?: string | null; force?: boolean; work?: boolean } = {}): Promise<{ config: boolean; memory: boolean; cookies: boolean; warnings: string[] }> {
return serialized(id, async () => {
const l = link(id);
l.assertOnline();
Expand Down Expand Up @@ -784,7 +785,11 @@ export function syncRunner(id: string, opts: { agentId?: string; profileId?: str
}
} catch (err) {
update("runners", id, { sync_error: `Couldn't copy your setup: ${message(err)}`, updated_at: now() });
throw Object.assign(new HttpError(502, `Couldn't copy your setup to ${r.name}: ${message(err)}`, "runner_sync_failed"), { cause: err });
const failure = Object.assign(new HttpError(502, `Couldn't copy your setup to ${r.name}: ${message(err)}`, "runner_sync_failed"), { cause: err });
// Work goes ahead on the setup the runner has (the runner card shows why) — not when it never got one or lacks the key.
if (!opts.work || needsKey || !r.synced_at || linkGone(err)) throw failure;
warnings.push(failure.message);
logFailure(failure.message, err);
} finally {
syncing.delete(id);
emit(id);
Expand Down Expand Up @@ -926,7 +931,7 @@ export async function startRemoteChat(runnerId: string, input: StartChatInput):
profileId = null;
}
}
await syncRunner(runnerId, { agentId, profileId });
await syncRunner(runnerId, { agentId, profileId, work: true });
const { runnerId: _runner, workingDirectory, computerTarget, vmId, androidDeviceIds, ...rest } = input;
// Another computer of the group is where the app that asked runs: its folders, screen, VMs and phones are named there.
const own = row(runnerId)?.kind === "device" ? { workingDirectory, computerTarget, vmId, androidDeviceIds } : {};
Expand All @@ -949,7 +954,7 @@ export async function prepareRemoteMessage(conversationId: string): Promise<void
profileId = null;
}
}
await syncRunner(runnerId, { agentId: conv.agentId, profileId });
await syncRunner(runnerId, { agentId: conv.agentId, profileId, work: true });
}

/* ------------------------------------------------------------------ */
Expand Down
49 changes: 48 additions & 1 deletion packages/core/src/remote/snapshot.ts
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ import { DEFAULT_SETTINGS, getSettings, resetSettingsCache } from "../services/s
import { badRequest, parseJson } from "../util";
import { decrypt, sha256 } from "../vault/crypto";
import { isSecondary, primaryRunnerId } from "./group";
import { asPeer, madeHere, pendingKeys } from "./replica";
import { asOwn, asPeer, madeHere, pendingKeys } from "./replica";
import * as vault from "../vault/vault";

const log = logger("snapshot");
Expand Down Expand Up @@ -222,6 +222,34 @@ export function appliedDigest(): string | null {
/* Apply (runner) */
/* ------------------------------------------------------------------ */

/**
* Tables with a unique slug besides the id, and the columns the slug is unique within. Two computers of a group can
* each make "earshot" in the same workspace while apart: the copy that comes second gets "earshot-2" instead of failing.
*/
export const SLUG_SCOPES: Record<string, string[]> = { workspaces: [], projects: ["workspace_id"] };

const SLUG_SEP = "\u001f";

export function slugKey(table: string, row: Record<string, unknown>, slug: unknown = row.slug): string {
return [...SLUG_SCOPES[table]!.map((c) => String(row[c] ?? "")), String(slug)].join(SLUG_SEP);
}

/** The id of another row of `table` that holds `row`'s slug. */
export function slugHolder(table: string, row: Record<string, unknown>, id: string, slug: unknown = row.slug): string | null {
const scope = SLUG_SCOPES[table]!;
const where = [...scope, "slug"].map((c) => `${q(c)} IS ?`).join(" AND ");
const params = [...scope.map((c) => (row[c] ?? null) as Value), slug as Value];
return get<{ id: string }>(`SELECT id FROM ${q(table)} WHERE ${where} AND id != ?`, ...params, id)?.id ?? null;
}

/** `row`'s slug, or the first "slug-2", "slug-3"… no other row holds and `claimed` doesn't name. */
export function freeSlug(table: string, row: Record<string, unknown>, id: string, claimed: Set<string> = new Set()): string {
const base = String(row.slug);
let candidate = base;
for (let i = 2; claimed.has(slugKey(table, row, candidate)) || slugHolder(table, row, id, candidate); i++) candidate = `${base}-${i}`;
return candidate;
}

/** Why a row can't be copied: its slug or id becomes a folder name on the runner (an agent's id does too, for its browser files). */
export function unsafe(table: string, row: Record<string, unknown>): string | null {
const label = typeof row.name === "string" && row.name ? row.name : String(row.id);
Expand Down Expand Up @@ -358,6 +386,25 @@ async function apply(s: ConfigSnapshot): Promise<{ digest: string; warnings: str
}
}

if (SLUG_SCOPES[spec.table] && pk === "id") {
const written = new Set(rows.map((r) => String(r.id)).filter((k) => !ownChanges.has(k)));
const claimed = new Map(rows.filter((r) => written.has(String(r.id))).map((r) => [slugKey(spec.table, r), String(r.id)]));
for (const local of all<Record<string, unknown>>(`SELECT * FROM ${q(spec.table)}`)) {
const key = String(local.id);
const holder = claimed.get(slugKey(spec.table, local));
if (!holder || holder === key) continue;
if (written.has(key)) {
// It gets its new slug from this copy too, maybe later in the loop: out of the way until then.
run(`UPDATE ${q(spec.table)} SET slug = ? WHERE id = ?`, `${String(local.slug)}~${key}`, key);
continue;
}
// Made here and not with the primary yet: it yields, and the new slug goes to the primary as this computer's change.
const slug = freeSlug(spec.table, local, key, new Set(claimed.keys()));
asOwn(() => run(`UPDATE ${q(spec.table)} SET slug = ? WHERE id = ?`, slug, key));
touch(spec, key);
}
}

for (const row of rows) {
const key = String(row[pk]);
if (ownChanges.has(key)) {
Expand Down
48 changes: 48 additions & 0 deletions packages/core/test/remote-snapshot.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@ import { setLogLevel } from "../src/log";
import type { MemorySnapshot } from "../src/memory/files";
import { mergeMemory, readMemoryState, writeMemoryState } from "../src/remote/memorySync";
import { appliedDigest, applySnapshot, buildSnapshot, snapshotDigest, type ConfigSnapshot } from "../src/remote/snapshot";
import { clearGroup, setGroup } from "../src/remote/group";
import { ensureReplica, madeHere, pendingKeys, resetReplicaCache } from "../src/remote/replica";
import * as runner from "../src/runner/runner";
import { getSettings, resetSettingsCache, updateSettings } from "../src/services/settings";
import { HttpError } from "../src/util";
Expand Down Expand Up @@ -613,6 +615,52 @@ describe("applying a snapshot on a runner", () => {
expect(get("SELECT 1 FROM settings WHERE key = 'future'")).toBeNull();
});

test("projects that swapped slugs on the controller swap on the runner too", async () => {
await sync(() => {
insert("workspaces", { id: "ws_apps", name: "Apps", slug: "apps", created_at: TS, updated_at: TS });
insert("projects", { id: "prj_one", workspace_id: "ws_apps", name: "One", slug: "alpha", created_at: TS, updated_at: TS });
insert("projects", { id: "prj_two", workspace_id: "ws_apps", name: "Two", slug: "beta", created_at: TS, updated_at: TS });
});
const result = await sync(() => {
sql("UPDATE projects SET slug = 'swap' WHERE id = 'prj_one'");
sql("UPDATE projects SET slug = 'alpha' WHERE id = 'prj_two'");
sql("UPDATE projects SET slug = 'beta' WHERE id = 'prj_one'");
});
expect(result.warnings).toEqual([]);
expect(row("projects", "prj_one").slug).toBe("beta");
expect(row("projects", "prj_two").slug).toBe("alpha");
});

test("a project a secondary made while apart yields its slug to the primary's and still goes to the primary", async () => {
const secondary = mkdtempSync(join(tmpdir(), "godmode-snapshot-secondary-"));
try {
use(secondary);
setGroup("secondary", "rnr_primary");
resetReplicaCache();
ensureReplica();
use(controller);
const before = buildSnapshot();
use(secondary);
await applySnapshot(before);
insert("projects", { id: "prj_mine", workspace_id: "ws_apps", name: "Earshot", slug: "earshot", created_at: LATER, updated_at: LATER });
use(controller);
insert("projects", { id: "prj_theirs", workspace_id: "ws_apps", name: "Earshot", slug: "earshot", created_at: LATER, updated_at: LATER });
const snapshot = buildSnapshot();
use(secondary);
expect((await applySnapshot(snapshot)).warnings).toEqual([]);
expect(row("projects", "prj_theirs").slug).toBe("earshot");
expect(row("projects", "prj_mine").slug).toBe("earshot-2");
expect(pendingKeys("rnr_primary", "projects").has("prj_mine")).toBe(true);
expect(madeHere("projects").has("prj_mine")).toBe(true);
} finally {
clearGroup();
resetReplicaCache();
use(controller);
sql("DELETE FROM projects WHERE id = 'prj_theirs'");
rmSync(secondary, { recursive: true, force: true });
}
});

test("unsafe ids are skipped with a warning and nothing is written outside the data directory", async () => {
use(controller);
const snapshot = buildSnapshot();
Expand Down
24 changes: 24 additions & 0 deletions packages/core/test/replica.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -248,6 +248,30 @@ describe("between two computers", () => {
expect(pendingFor("rnr_b")).toBe(0);
});

test("a project both computers made while apart lands with its own slug instead of failing", () => {
use(b);
run("DELETE FROM projects WHERE workspace_id = 'wsp_apps'");
run("DELETE FROM workspaces WHERE id = 'wsp_apps'");
run("INSERT INTO workspaces (id, name, slug, created_at, updated_at) VALUES ('wsp_apps', 'Apps', 'apps', ?, ?)", T(0), T(0));
run("INSERT INTO projects (id, workspace_id, name, slug, created_at, updated_at) VALUES ('prj_here', 'wsp_apps', 'Earshot', 'earshot', ?, ?)", T(1), T(1));
const theirs: ReplicaChange = {
t: "projects",
k: "prj_there",
at: T(30),
row: { id: "prj_there", workspace_id: "wsp_apps", name: "Earshot", slug: "earshot", created_at: T(30), updated_at: T(30) },
};
// Within the pull the slug may still be freed by a rename that comes later: it waits for the end of the pull.
expect(applyChanges("rnr_a", [theirs], { peerWinsTies: false }).failed).toBe(1);
const res = retryDeferred("rnr_a", { peerWinsTies: false });
expect(res.applied).toBe(1);
expect(res.failed).toBe(0);
expect(all("SELECT id, slug FROM projects WHERE workspace_id = 'wsp_apps' ORDER BY id")).toEqual([
{ id: "prj_here", slug: "earshot" },
{ id: "prj_there", slug: "earshot-2" },
]);
expect(all("SELECT * FROM replica_deferred")).toHaveLength(0);
});

test("an agent whose folder name isn't safe never lands", () => {
use(b);
const evil: ReplicaChange = { t: "agents", k: "agt_evil", at: T(20), row: { id: "agt_evil", name: "Evil", slug: "../../outside", created_at: T(0), updated_at: T(0) } };
Expand Down
Loading