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
46 changes: 33 additions & 13 deletions control/src/server/relay-control.ts
Original file line number Diff line number Diff line change
Expand Up @@ -161,6 +161,32 @@ export class RelayControl {
throw new ControlError("state_publication_failed", 503);
}
}
private mutate<T>(change: () => T): T {
const revision = this.reserveRevision();
this.healthy = false;
try {
const result = this.db.transaction(() => {
const result = change();
this.publish(revision);
return result;
})();
// Publication alone cannot establish readiness: COMMIT may still fail.
this.healthy = true;
return result;
} catch (error) {
this.healthy = false;
if (!this.db.inTransaction && this.publicationFloor < revision) {
// Validation failed before publication. Consume the reserved revision
// by publishing the rolled-back, committed state, not the rejected
// mutation. Do not leave ordinary 4xx requests poisoning readiness.
try { this.publish(revision); } catch { /* remain fail-closed */ }
}
// If publication started, its revision may already be visible. Keep
// admission/readiness closed until periodic publication or restart
// repairs the snapshot from committed DB state at a higher revision.
throw error;
}
}
publish(reservedRevision?: number) {
const revision = reservedRevision ?? this.reserveRevision();
const current = this.db.prepare("SELECT revision FROM relay_public_revision WHERE id=1").get() as { revision: number };
Expand Down Expand Up @@ -212,7 +238,7 @@ export class RelayControl {
} finally {
closeSync(dirfd);
}
this.healthy = true;
this.healthy = !this.db.inTransaction;
this.publishedState = JSON.stringify(snapshot);
this.publishedAt = now;
} catch (e) {
Expand Down Expand Up @@ -473,8 +499,7 @@ export class RelayControl {
const existing = this.operationRecord(userId, p);
assert(existing, "operation_not_found", 404);
if (existing.result !== null) return JSON.parse(existing.result);
const revision = this.reserveRevision();
return this.db.transaction(() => {
return this.mutate(() => {
const row = this.operationRecord(userId, p);
assert(row, "operation_not_found", 404);
// Historical receipt only: no new mutation, proof reexecution or revival.
Expand Down Expand Up @@ -545,9 +570,8 @@ export class RelayControl {
"UPDATE relay_operations SET result=? WHERE operationId=? AND userId=?",
)
.run(JSON.stringify(result), p.operationId, userId);
this.publish(revision);
return result;
})();
});
}
recover(userId: string, input: unknown) {
assert(this.healthy, "state_unavailable", 503);
Expand All @@ -561,8 +585,7 @@ export class RelayControl {
const existing = this.operationRecord(userId, p);
assert(existing, "operation_not_found", 404);
if (existing.result !== null) return JSON.parse(existing.result);
const revision = this.reserveRevision();
return this.db.transaction(() => {
return this.mutate(() => {
const row = this.operationRecord(userId, p);
assert(row, "operation_not_found", 404);
if (row.result !== null) return JSON.parse(row.result);
Expand All @@ -582,9 +605,8 @@ export class RelayControl {
this.db.prepare(
"UPDATE relay_operations SET result=? WHERE operationId=? AND userId=?",
).run(JSON.stringify(result), p.operationId, userId);
this.publish(revision);
return result;
})();
});
}
async admit(userId: string, input: unknown) {
assert(this.healthy, "state_unavailable", 503);
Expand Down Expand Up @@ -634,8 +656,7 @@ export class RelayControl {
principal(id);
assert(this.healthy, "state_unavailable", 503);
this.device(id, userId, false);
const revision = this.reserveRevision();
return this.db.transaction(() => {
return this.mutate(() => {
this.device(id, userId, false);
this.db
.prepare(
Expand All @@ -645,8 +666,7 @@ export class RelayControl {
this.db
.prepare("DELETE FROM relay_challenges WHERE userId=?")
.run(userId);
this.publish(revision);
return { principal: id, revoked: true };
})();
});
}
}
142 changes: 142 additions & 0 deletions control/tests/mutation-consistency.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,142 @@
import { afterEach, expect, it } from "vitest";
import { randomUUID } from "node:crypto";
import { readFileSync, renameSync } from "node:fs";
import { join } from "node:path";
import { fixture, identity } from "./fixtures.js";
import { RelayControl } from "../src/server/relay-control.js";
import { openDatabase } from "../src/server/storage.js";

let f: Awaited<ReturnType<typeof fixture>> | undefined;
afterEach(() => { f?.cleanup(); f = undefined; });
const state = () => JSON.parse(readFileSync(join(f!.config.publicDir, "state.json"), "utf8"));
async function expectHealthy() {
expect(f!.control.isHealthy()).toBe(true);
expect((await f!.request("/healthz")).status).toBe(200);
expect(f!.db.prepare("SELECT revision FROM relay_public_revision WHERE id=1").get())
.toEqual({ revision: state().revision });
}

it("keeps readiness after invalid registration proof and allows the same challenge to succeed", async () => {
f = await fixture();
const d = identity(f.root, "device", "a");
const request = f.proven(f.owner.id, "register", d.payload, d.privateKey);
const before = state().revision;
const response = await f.request("/api/relay/devices", { ...request, proof: "bad" });
expect(response.status).toBe(400);
await expectHealthy();
expect(state().revision).toBeGreaterThan(before);
expect(state().devices).toEqual({});
expect(f.control.operation(f.owner.id, d.payload.operationId).committed).toBe(false);
const receipt = f.control.register(f.owner.id, request);
const committedRevision = state().revision;
expect(f.control.register(f.owner.id, request)).toEqual(receipt);
expect(state().revision).toBe(committedRevision);
await expectHealthy();
});

it("keeps readiness when a competing rotation wins before a stale request", async () => {
f = await fixture();
const old = identity(f.root, "old", "a");
const next = identity(f.root, "next", "b");
f.control.register(f.owner.id, f.proven(f.owner.id, "register", old.payload, old.privateKey));
const payload = { ...old.payload, certificatePEM: next.payload.certificatePEM,
expectedGeneration: 0, keyGeneration: 1, operationId: randomUUID() };
const losing = f.rotated(f.owner.id, payload, old.privateKey, next.privateKey);
const winning = f.rotated(f.owner.id, { ...payload, operationId: randomUUID() }, old.privateKey, next.privateKey);
f.control.register(f.owner.id, winning);
const committed = state();
expect(() => f!.control.register(f!.owner.id, losing)).toThrow("generation_conflict");
await expectHealthy();
expect(state().devices).toEqual(committed.devices);
expect(state().revision).toBeGreaterThan(committed.revision);
expect(f.control.operation(f.owner.id, payload.operationId).committed).toBe(false);
});

it.each(["register", "recover", "revoke"] as const)("fails closed on %s COMMIT failure, then republishes only rolled-back DB state", async (kind) => {
f = await fixture();
const old = identity(f.root, "old", "a");
f.control.register(f.owner.id, f.proven(f.owner.id, "register", old.payload, old.privateKey));
const next = identity(f.root, "next", "b");
const payload = kind === "recover"
? { ...next.payload, oldPrincipal: old.payload.principal }
: next.payload;
const request = kind === "revoke" ? undefined : f.proven(f.owner.id, kind, payload, next.privateKey);
const run = () => kind === "revoke" ? f!.control.revoke(f!.owner.id, old.payload.principal)
: kind === "register" ? f!.control.register(f!.owner.id, request)
: f!.control.recover(f!.owner.id, request);
const before = state();
// A deferred FK passes the mutation/publication and fails only at COMMIT.
f.db.exec(`CREATE TABLE fault_parent(id TEXT PRIMARY KEY);
CREATE TABLE fault_child(id TEXT REFERENCES fault_parent(id) DEFERRABLE INITIALLY DEFERRED);
CREATE TRIGGER fault_commit AFTER ${kind === "register" ? "INSERT" : "UPDATE OF revoked"} ON relay_devices
BEGIN INSERT INTO fault_child VALUES ('missing'); END;`);
expect(run).toThrow("FOREIGN KEY constraint failed");
const exposed = state();
expect(exposed.revision).toBeGreaterThan(before.revision);
expect(f.control.list(f.owner.id)).toMatchObject([{ principal: old.payload.principal, revoked: false }]);
if (kind !== "revoke") expect(f.control.operation(f.owner.id, payload.operationId).committed).toBe(false);
expect(f.control.healthy).toBe(false);
expect((await f.request("/healthz")).status).toBe(503);
const admission = { role: "receiver", devicePrincipal: old.payload.principal,
receiverPrincipal: old.payload.principal };
await expect(f.control.admit(f.owner.id, admission)).rejects.toThrow("state_unavailable");
f.db.exec("DROP TRIGGER fault_commit");
if (kind === "revoke") {
// Startup can repair a failed COMMIT before the periodic timer runs.
const recoveryDb = openDatabase(f.config.dataDir);
try {
const restarted = new RelayControl(recoveryDb, f.config);
expect(restarted.isHealthy()).toBe(true);
expect(state().revision).toBeGreaterThan(exposed.revision);
expect(state().devices).toEqual(before.devices);
} finally { recoveryDb.close(); }
}
// The normal periodic publication repairs the snapshot with a higher revision.
f.control.publish();
expect(state().revision).toBeGreaterThan(exposed.revision);
expect(state().devices).toEqual(before.devices);
await expectHealthy();
const receipt = run();
if (kind !== "revoke") expect(run()).toEqual(receipt);
await expectHealthy();
const second = openDatabase(f.config.dataDir);
try {
const restarted = new RelayControl(second, f.config);
expect(restarted.isHealthy()).toBe(true);
expect(restarted.list(f.owner.id)).toEqual(f.control.list(f.owner.id));
if (kind !== "revoke") expect(restarted.operation(f.owner.id, payload.operationId)).toEqual(receipt);
} finally { second.close(); }
});

it("keeps readiness after invalid recovery proof without fencing either device", async () => {
f = await fixture();
const old = identity(f.root, "old", "a");
const next = identity(f.root, "next", "b");
f.control.register(f.owner.id, f.proven(f.owner.id, "register", old.payload, old.privateKey));
const request = f.proven(f.owner.id, "recover", { ...next.payload, oldPrincipal: old.payload.principal }, next.privateKey);
const before = state();
expect(() => f!.control.recover(f!.owner.id, { ...request, proof: "bad" })).toThrow("invalid_proof");
await expectHealthy();
expect(state().devices).toEqual(before.devices);
f.control.recover(f.owner.id, request);
await expectHealthy();
});

it("stays fail-closed if rejection repair cannot publish, then recovers without reusing a revision", async () => {
f = await fixture();
const d = identity(f.root, "device", "a");
const request = f.proven(f.owner.id, "register", d.payload, d.privateKey);
const before = state().revision;
renameSync(f.config.publicDir, f.config.publicDir + "-away");
expect(() => f!.control.register(f!.owner.id, { ...request, proof: "bad" })).toThrow("invalid_proof");
expect(f.control.healthy).toBe(false);
expect((await f.request("/healthz")).status).toBe(503);
expect(() => f!.control.register(f!.owner.id, request)).toThrow("state_unavailable");
renameSync(f.config.publicDir + "-away", f.config.publicDir);
f.control.publish();
expect(state().revision).toBeGreaterThan(before + 1);
expect(state().devices).toEqual({});
await expectHealthy();
f.control.register(f.owner.id, request);
await expectHealthy();
});
2 changes: 1 addition & 1 deletion docs/i18n-manifest.json
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@
"docs/relay-google-bootstrap.md": "bf6ce95c1ba39a56b765542d90e49c24211de5dd7cecb453819dfc35388bed9d",
"docs/relay-lifecycle.md": "ab204780ce0b9a9b809e72bf20370a824e0292068218e98a1ad5449f30bcfa70",
"docs/relay-public-pilot-2026-09-17.md": "e86c5528533178663f38662e15f1d7a12b6db1eab9a722e3bdbd29dd033bc1ac",
"docs/relay-watchdog.md": "bccb789ed27e3ec5ae36343f91b2d2710d6bb3c4c8735353a07d98c8f258b5ae",
"docs/relay-watchdog.md": "f37fc1fae8627e7a58e2a0385cf45e7161ed123a237c451857d23a03c8ebec6f",
"docs/releases/v0.6.1.md": "7fae8ab4498228d8498518c1de20de857271ae14089fe9e0bf6f7af828bd77fc",
"docs/releases/v0.6.2.md": "ca3a87de507c6c247a6bd0fe1010d18eee164cfede1e643e9f32d2d7bc87dcf8",
"docs/releases/v0.7.0.md": "fa1081169a302c596246a29ea7a1a9c48aa697f586bf24740e9ebe5541970d74",
Expand Down
2 changes: 2 additions & 0 deletions docs/ja/relay-watchdog.md
Original file line number Diff line number Diff line change
Expand Up @@ -22,3 +22,5 @@ HTTP の `/healthz` エンドポイントでも同じ状態チェックが使用
ローカル単体テストは、通知ゲーティング、障害、有界な並行性、環境の分離、および状態ヘルスチェックをカバーしています。これらは実際の systemd 通知配信や SIGSTOP 復旧ゲートを確立するものではなく、それには KR ランタイム実験が必要です。

参考資料: [systemd v255 通知セマンティクス](https://github.com/systemd/systemd/blob/v255/man/systemd-notify.xml) および [サービスウォッチドッグオプション](https://github.com/systemd/systemd/blob/v255/man/systemd.service.xml)。

デバイス変更はトランザクション前にリビジョンを永続予約し、再利用しません。公開前の検証で要求を拒否した場合、Control は予約したリビジョンでコミット済み状態を再公開し、拒否だけで readiness が失われないようにします。トランザクション内の公開は COMMIT が戻るまで成功ではありません。公開または COMMIT の失敗時は readiness と admission を閉じ、定期公開(60秒間隔)または再起動でコミット済み DB 状態をより高いリビジョンに正常公開するまで維持します。SQLite と公開ファイルは単一の原子的資源ではなく、未コミットのスナップショットが一時的に見えるため、協調スタックと watchdog は引き続き必要です。ストレージ障害が続けば unhealthy を維持します。リビジョンカウンターや受領証を削除しないでください。この修正は Python パッケージだけでなく、別途デプロイする Control サービスのものです。
2 changes: 2 additions & 0 deletions docs/ko/relay-watchdog.md
Original file line number Diff line number Diff line change
Expand Up @@ -22,3 +22,5 @@ HTTP `/healthz` 엔드포인트도 동일한 상태 검사를 사용합니다.
로컬 단위 테스트는 알림 게이팅, 실패, 유계 동시성(bounded concurrency), 환경 격리 및 상태 헬스 체크를 다룹니다. 이는 실제 systemd 알림 전달이나 SIGSTOP 복구 게이트를 확립하지 않으며, 이를 위해서는 KR 런타임 실험이 필요합니다.

참고자료: [systemd v255 알림 시맨틱](https://github.com/systemd/systemd/blob/v255/man/systemd-notify.xml) 및 [서비스 워치독 옵션](https://github.com/systemd/systemd/blob/v255/man/systemd.service.xml).

기기 변경은 트랜잭션 전에 revision을 영구 예약하며 재사용하지 않습니다. 게시 전 검증에서 요청이 거절되면 Control은 커밋된 상태를 예약된 revision으로 다시 게시해 거절만으로 readiness가 저하되지 않도록 합니다. 트랜잭션 안의 게시는 COMMIT 반환 전까지 성공이 아닙니다. 게시 또는 COMMIT 실패 시 readiness와 admission을 차단하며, 주기적 게시(60초 간격)나 재시작으로 커밋된 DB 상태를 더 높은 revision에 성공적으로 게시한 뒤 복구합니다. SQLite와 공개 파일은 하나의 원자적 자원이 아니므로 미커밋 스냅샷이 잠시 보일 수 있으며, 연동된 스택과 watchdog이 계속 필요합니다. 저장소 장애가 지속되면 unhealthy를 유지합니다. revision 카운터나 영수증을 삭제하지 마십시오. 이 수정은 Python 패키지뿐 아니라 별도 배포되는 Control 서비스에 해당합니다.
2 changes: 2 additions & 0 deletions docs/relay-watchdog.md
Original file line number Diff line number Diff line change
Expand Up @@ -56,3 +56,5 @@ delivery or the SIGSTOP recovery gate; that requires the KR runtime experiment.

References: [systemd v255 notification semantics](https://github.com/systemd/systemd/blob/v255/man/systemd-notify.xml)
and [service watchdog options](https://github.com/systemd/systemd/blob/v255/man/systemd.service.xml).

Device mutations reserve revisions durably before their transaction; revisions are never reused. If validation rejects a request before publication, Control republishes committed state at the reserved revision so rejection alone does not poison readiness. Publication inside a transaction is not success until COMMIT returns. Publication or COMMIT failure keeps readiness and admission closed until a successful periodic publication (60-second interval) or restart publishes committed DB state at a higher revision. SQLite and the public file are not one atomic resource: an uncommitted snapshot can be briefly visible, and the coordinated stack/watchdog remains required. Persistent storage failures stay unhealthy; do not delete the revision counter or receipts. This fix belongs to the separately deployed Control service, not only the Python package.
2 changes: 2 additions & 0 deletions docs/zh-CN/relay-watchdog.md
Original file line number Diff line number Diff line change
Expand Up @@ -22,3 +22,5 @@ HTTP `/healthz` 端点使用相同的状态检查。读取失败、状态缺失
本地单元测试涵盖通知门控、失败、有界并发、环境隔离和状态健康检查。它们并不建立实际的 systemd 通知投递或 SIGSTOP 恢复门控;这需要 KR 运行时实验。

参考资料:[systemd v255 通知语义](https://github.com/systemd/systemd/blob/v255/man/systemd-notify.xml) 以及 [服务看门狗选项](https://github.com/systemd/systemd/blob/v255/man/systemd.service.xml)。

设备变更在事务前持久预留修订号,且绝不复用。若发布前验证拒绝请求,Control 会以预留修订号重新发布已提交状态,避免仅因拒绝请求而损害 readiness。事务内发布在 COMMIT 返回前不算成功。发布或 COMMIT 失败会关闭 readiness 和 admission,直到周期发布(60秒间隔)或重启以更高修订号成功发布已提交 DB 状态。SQLite 与公开文件不是一个原子资源:未提交快照可能短暂可见,因此仍需协调的服务栈和 watchdog。持续存储故障会保持 unhealthy;不要删除修订号计数器或收据。本修复属于独立部署的 Control 服务,而不只是 Python 包。
Loading