Skip to content

fix(workstreams): recover interrupted member key initialization #941

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
Mjones13 merged 1 commit into from
Nov 12, 2026
Merged

fix(workstreams): recover interrupted member key initialization #941

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
135 changes: 103 additions & 32 deletions apps/server/src/jones/workstreams/native/cli.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -203,18 +203,27 @@ export const keyLayer = Layer.effect(
}),
);

interface MemberKeyRow {
readonly key_id: string;
readonly registry_instance_id: string;
readonly scope: string;
readonly state: string;
readonly created_at: string;
}

export class NativeMemberKeyAdmin extends Context.Service<
NativeMemberKeyAdmin,
{
readonly initialize: (
input: MemberKeyInit,
options?: { readonly resume: boolean },
) => Effect.Effect<{ readonly created: readonly string[] }, RegistryError>;
}
>()("t3/jones/workstreams/native/cli/NativeMemberKeyAdmin") {}
/**
* A member holds no registry copy, only the active `local` key row (no FK to a registry row) and its
* key file for the home's registry id. Any registry row or key row for that id, or an existing file,
* refuses. The row write and the readback share one transaction; a failed run removes only its file.
* key file for the home's registry id. Resume explicitly adopts an orphan file or an exact active row.
* A failed transaction may have committed: reconcile outside it before removing an owned file.
*/
export const memberKeyLayer = Layer.effect(
NativeMemberKeyAdmin,
Expand All @@ -224,56 +233,110 @@ export const memberKeyLayer = Layer.effect(
sql = yield* SqlClient.SqlClient;
const refuse = (code: string) => (cause: unknown) => new RegistryError({ code, cause });
return NativeMemberKeyAdmin.of({
initialize: Effect.fn("NativeMemberKeyAdmin.initialize")(function* (input) {
initialize: Effect.fn("NativeMemberKeyAdmin.initialize")(function* (input, options) {
yield* decodeMemberKeyInit(input).pipe(Effect.mapError(refuse("invalid_request")));
const key_id = input.local_key_id,
registry_instance_id = input.registry.registry_instance_id,
name = `workstreams-content-key-${key_id}`;
let created = false;
yield* sql
let created: Uint8Array | undefined;
let insertedAt: string | undefined;
const readRows = Effect.fn("NativeMemberKeyAdmin.readRows")(
() =>
sql<MemberKeyRow>`SELECT key_id, registry_instance_id, scope, state, created_at FROM jones_workstreams_keys
WHERE key_id = ${key_id} OR registry_instance_id = ${registry_instance_id}`,
);
const exactRow = (rows: ReadonlyArray<MemberKeyRow>) =>
rows.length === 1 &&
rows[0]!.key_id === key_id &&
rows[0]!.registry_instance_id === registry_instance_id &&
rows[0]!.scope === "local" &&
rows[0]!.state === "active";
const sameBytes = (a: Uint8Array, b: Uint8Array) =>
a.byteLength === b.byteLength && a.every((byte, index) => byte === b[index]);
return yield* sql
.withTransaction(
Effect.gen(function* () {
const [existing] = yield* sql<{ registries: number; keys: number }>`SELECT
(SELECT count(*) FROM jones_workstreams_registries WHERE registry_instance_id = ${registry_instance_id}) AS registries,
(SELECT count(*) FROM jones_workstreams_keys WHERE key_id = ${key_id} OR registry_instance_id = ${registry_instance_id}) AS keys`;
const [existing] = yield* sql<{ registries: number }>`SELECT
count(*) AS registries FROM jones_workstreams_registries WHERE registry_instance_id = ${registry_instance_id}`;
if (existing!.registries !== 0)
return yield* new RegistryError({ code: "already_initialized" });
if (existing!.keys !== 0) return yield* new RegistryError({ code: "key_exists" });
if (
Option.isSome(yield* secrets.get(name).pipe(Effect.mapError(refuse("unavailable"))))
)
const rows = yield* readRows();
const file = yield* secrets.get(name).pipe(Effect.mapError(refuse("unavailable")));
if (rows.some((row) => row.key_id === key_id) && Option.isNone(file))
return yield* new RegistryError({ code: "key_file_missing" });
if (rows.length !== 0) {
if (!options?.resume || !exactRow(rows))
return yield* new RegistryError({ code: "key_exists" });
if (Option.isNone(file))
return yield* new RegistryError({ code: "key_file_missing" });
if (file.value.byteLength !== 32)
return yield* new RegistryError({ code: "invalid_key" });
return { created: [] };
}
if (Option.isSome(file) && !options?.resume)
return yield* new RegistryError({ code: "key_exists" });
const bytes = yield* crypto
.randomBytes(32)
.pipe(Effect.mapError(refuse("unavailable")));
yield* secrets
.create(name, bytes)
.pipe(
Effect.mapError((cause) =>
refuse(
SecretStore.isSecretAlreadyExistsError(cause) ? "key_exists" : "unavailable",
)(cause),
),
);
created = true;
if (Option.isSome(file) && file.value.byteLength !== 32)
return yield* new RegistryError({ code: "invalid_key" });
if (Option.isNone(file)) {
const bytes = yield* crypto
.randomBytes(32)
.pipe(Effect.mapError(refuse("unavailable")));
yield* secrets
.create(name, bytes)
.pipe(
Effect.mapError((cause) =>
refuse(
SecretStore.isSecretAlreadyExistsError(cause)
? "key_exists"
: "unavailable",
)(cause),
),
);
created = bytes;
}
const at = DateTime.formatIso(yield* DateTime.now);
insertedAt = at;
yield* sql`INSERT INTO jones_workstreams_keys ${sql.insert({ key_id, registry_instance_id, scope: "local", state: "active", created_at: at })}`;
// The no-replace trigger ignores a duplicate key id silently; read the row back.
const rows = yield* sql<{ key_id: string }>`SELECT key_id FROM jones_workstreams_keys
WHERE registry_instance_id = ${registry_instance_id} AND scope = 'local' AND state = 'active'`;
if (rows.length !== 1 || rows[0]!.key_id !== key_id)
const written = yield* readRows();
if (!exactRow(written) || written[0]!.created_at !== at)
return yield* new RegistryError({ code: "unavailable" });
return { created: created ? [name] : [] };
}),
)
.pipe(
Effect.mapError((cause) =>
isRegistryError(cause) ? cause : new RegistryError({ code: "unavailable", cause }),
),
Effect.tapError(() =>
created ? secrets.remove(name).pipe(Effect.ignore) : Effect.void,
Effect.catch((error) =>
Effect.gen(function* () {
if (!created) return yield* error;
const readback = yield* Effect.result(readRows());
if (readback._tag === "Failure")
return yield* new RegistryError({
code: "outcome_unknown",
cause: readback.failure,
});
if (
exactRow(readback.success) &&
insertedAt !== undefined &&
readback.success[0]!.created_at === insertedAt
)
return { created: [name] };
if (readback.success.length !== 0)
return yield* new RegistryError({ code: "outcome_unknown", cause: error });
const file = yield* secrets
.get(name)
.pipe(Effect.mapError(refuse("outcome_unknown")));
if (Option.isSome(file)) {
if (!sameBytes(file.value, created))
return yield* new RegistryError({ code: "outcome_unknown", cause: error });
yield* secrets.remove(name).pipe(Effect.mapError(refuse("outcome_unknown")));
}
return yield* error;
}),
),
);
return { created: [name] };
}),
});
}),
Expand Down Expand Up @@ -331,6 +394,12 @@ export const registryCommand = <E, R, SE, SR, TE, TR, KE, KR, ME, MR>(
Command.make("member-init", {
home: Flag.String("home"),
keyProvider: Flag.String("key-provider"),
resume: Flag.Boolean("resume").pipe(
Flag.withDefault(false),
Flag.withDescription(
"Reuse a valid orphan key or accept an exact existing member key and row.",
),
),
request: Flag.String("request").pipe(
Flag.withDescription(
"JSON {registry, local_key_id}: the home's registry and this member's local key id.",
Expand All @@ -346,7 +415,9 @@ export const registryCommand = <E, R, SE, SR, TE, TR, KE, KR, ME, MR>(
onExcessProperty: "error",
});
const result = yield* Effect.gen(function* () {
return yield* (yield* NativeMemberKeyAdmin).initialize(input);
return yield* (yield* NativeMemberKeyAdmin).initialize(input, {
resume: flags.resume,
});
}).pipe(
Effect.provide(
memberKeyBootstrap({ home: flags.home, keyProvider: flags.keyProvider }),
Expand Down
63 changes: 33 additions & 30 deletions apps/server/src/jones/workstreams/native/conformance/peerSetupIntegration.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,6 +159,7 @@ import * as Bindings from "../store/bindingStore.ts";
import * as Database from "../store/sql.ts";
import * as Envelope from "../crypto/envelope.ts";
import * as Keys from "../crypto/keyProvider.ts";
import * as Admin from "../cli.ts";
import * as Owner from "../runtime/setupOwner.ts";
import * as Setup from "../runtime/peerSetup.ts";
import * as Files from "../runtime/peerSetupRequestFile.ts";
Expand Down Expand Up @@ -266,38 +267,23 @@ const fixture = Effect.gen(function* () {
);
NodeFS.mkdirSync(f.secretsDir, { mode: 0o700 });
for (const [name, key] of [
...Test.secrets(),
...[...Test.secrets()].filter(
([name]) => env === "home:test" || name !== "workstreams-content-key-test-local",
),
["server-signing-key", new Uint8Array(32).fill(env === "home:test" ? 41 : 51)] as const,
])
NodeFS.writeFileSync(NodePath.join(f.secretsDir, name + ".bin"), key, { mode: 0o600 });
}
const readSecret = (name: string) =>
controls.beforeSecret(env, name).pipe(
Effect.andThen(
Effect.sync(() => {
if (!/^[A-Za-z0-9._:-]+$/.test(name))
throw new Error("invalid synthetic secret name");
const path = NodePath.join(f.secretsDir, name + ".bin");
return NodeFS.existsSync(path)
? Option.some(new Uint8Array(NodeFS.readFileSync(path)))
: Option.none<Uint8Array>();
}),
),
);
const secrets = Layer.succeed(Secrets.ServerSecretStore, {
get: readSecret,
getOrCreateRandom: (name) =>
readSecret(name).pipe(
Effect.flatMap((value) =>
Option.isSome(value)
? Effect.succeed(value.value)
: Effect.die("missing existing fixture key"),
),
),
set: () => Effect.die("no fixture key mutation"),
create: () => Effect.die("no fixture key creation"),
remove: () => Effect.die("no fixture key removal"),
});
const secrets = Layer.effect(
Secrets.ServerSecretStore,
Effect.gen(function* () {
const store = yield* Secrets.make;
return Secrets.ServerSecretStore.of({
...store,
get: (name) => controls.beforeSecret(env, name).pipe(Effect.andThen(store.get(name))),
});
}),
);
const storage = Layer.mergeAll(
Registry.layer,
PinRecords.layer,
Expand Down Expand Up @@ -339,10 +325,10 @@ const fixture = Effect.gen(function* () {
yield* m129;
const sql = yield* SqlClient.SqlClient;
yield* sql`CREATE TABLE jones_workstreams_setup_writer_probe(value TEXT)`;
yield* sql`INSERT INTO jones_workstreams_keys(key_id,registry_instance_id,scope,state,created_at)
VALUES ('test-local',${home.registry_instance_id},'local','active',${Test.now})`;
if (env === "home:test") {
yield* sql`INSERT INTO jones_workstreams_keys(key_id,registry_instance_id,scope,state,created_at)
VALUES ('test-local',${home.registry_instance_id},'local','active',${Test.now})`;
yield* sql`INSERT INTO jones_workstreams_keys(key_id,registry_instance_id,scope,state,created_at)
VALUES ('test-home',${home.registry_instance_id},'home','active',${Test.now})`;
const registry = yield* Registry.RegistryStore;
yield* registry.initialize({
Expand All @@ -366,6 +352,23 @@ const fixture = Effect.gen(function* () {
permissions: JSON.stringify(["workstreams:read", "workstreams:write"]),
state: "active",
});
} else {
yield* Effect.gen(function* () {
yield* (yield* Admin.NativeMemberKeyAdmin).initialize({
registry: { ...home, canonical_origin: "https://native.workstreams.invalid" },
local_key_id: "test-local",
});
}).pipe(Effect.provide(Admin.memberKeyLayer));
const keys = yield* Keys.WorkstreamsKeyProvider;
const id = yield* keys.activeKeyId({ ...home, key_scope: "local" });
expect(id).toBe("test-local");
expect(yield* keys.get(id)).toEqual(
new Uint8Array(
NodeFS.readFileSync(
NodePath.join(f.secretsDir, "workstreams-content-key-test-local.bin"),
),
),
);
}
}).pipe(Effect.provide(context));
NodeFS.chmodSync(f.dbPath, 0o600);
Expand Down
Loading