457 lines
18 KiB
TypeScript
457 lines
18 KiB
TypeScript
import assert from "node:assert/strict";
|
|
import { createHash } from "node:crypto";
|
|
import { mkdtempSync, readFileSync, readdirSync, rmSync } from "node:fs";
|
|
import { tmpdir } from "node:os";
|
|
import { join } from "node:path";
|
|
import { describe, it } from "node:test";
|
|
|
|
import type { ElyR2Object, ElyR2PutOptions, Env } from "../src/bindings.js";
|
|
import { recentDeviceActionProofBytes } from "../src/recent_device_action_proof.js";
|
|
import {
|
|
SYNC_R2_ANONYMIZE_USER_QUERY,
|
|
SYNC_R2_FENCE_USER_QUERY,
|
|
abandonSyncR2Write,
|
|
claimSyncR2SnapshotWrite,
|
|
collectSyncR2Garbage,
|
|
} from "../src/sync_r2_gc.js";
|
|
import { inventorySyncR2Objects } from "../src/sync_r2_inventory.js";
|
|
import { syncResetDocument } from "../src/sync_reset.js";
|
|
import { PUBLIC_KEY, signDeviceMessage } from "./devices_test_support.js";
|
|
import { SqliteD1Database, execute, query } from "./sqlite_d1_test_support.js";
|
|
|
|
const USER_ID = "user-01";
|
|
const DEVICE_ID = "device-01";
|
|
const KEY_ID = "1".repeat(64);
|
|
const OWNER_HASH = createHash("sha256").update(USER_ID).digest("hex");
|
|
const HASH_A = "a".repeat(64);
|
|
const HASH_B = "b".repeat(64);
|
|
const TOKEN_A = "c".repeat(64);
|
|
const TOKEN_B = "d".repeat(64);
|
|
const NOW = 1_800_000_000;
|
|
const MIGRATIONS_DIR = join(process.cwd(), "migrations");
|
|
|
|
describe("sync R2 GC SQLite state machine", () => {
|
|
it("commits a leased candidate and deletes it only after D1 references are fenced", async () => {
|
|
await withDatabase(async (databasePath, database, bucket, env) => {
|
|
const key = snapshotKey(HASH_A);
|
|
const lease = await claimSyncR2SnapshotWrite(env, snapshotClaim(key), NOW, TOKEN_A);
|
|
await bucket.put(key, bytes("ciphertext-a"));
|
|
commitGenesis(databasePath, key, HASH_A, lease.writeToken);
|
|
|
|
assert.equal(candidateState(databasePath, key), "referenced");
|
|
assert.equal(await collectSyncR2Garbage(env, NOW + 100_000), 0);
|
|
|
|
await database.batch([
|
|
database.prepare(SYNC_R2_FENCE_USER_QUERY).bind(NOW + 1, NOW + 1, NOW + 1, USER_ID),
|
|
database.prepare("DELETE FROM sync_snapshot_heads WHERE user_id = ?").bind(USER_ID),
|
|
database.prepare("DELETE FROM sync_snapshot_encryption WHERE user_id = ?").bind(USER_ID),
|
|
database.prepare("DELETE FROM sync_snapshots WHERE user_id = ?").bind(USER_ID),
|
|
]);
|
|
assert.equal(candidateState(databasePath, key), "ready");
|
|
assert.equal(await collectSyncR2Garbage(env, NOW + 1), 1);
|
|
assert.equal(candidateState(databasePath, key), "deleted");
|
|
assert.deepEqual(bucket.deletes, [key]);
|
|
});
|
|
});
|
|
|
|
it("turns a CAS loser into an immediately collectible ready candidate", async () => {
|
|
await withDatabase(async (databasePath, _database, bucket, env) => {
|
|
const key = snapshotKey(HASH_A);
|
|
const lease = await claimSyncR2SnapshotWrite(env, snapshotClaim(key), NOW, TOKEN_A);
|
|
await bucket.put(key, bytes("ciphertext-a"));
|
|
|
|
await abandonSyncR2Write(env, USER_ID, OWNER_HASH, key, lease.writeToken, NOW + 1);
|
|
|
|
assert.equal(candidateState(databasePath, key), "ready");
|
|
assert.equal(await collectSyncR2Garbage(env, NOW + 1), 1);
|
|
assert.equal(candidateState(databasePath, key), "deleted");
|
|
assert.deepEqual(bucket.deletes, [key]);
|
|
});
|
|
});
|
|
|
|
it("resets sync state while preserving the vault generation and device envelopes", async () => {
|
|
await withDatabase(async (databasePath, _database, bucket, env) => {
|
|
seedVaultEnvelope(databasePath);
|
|
const key = snapshotKey(HASH_A);
|
|
const lease = await claimSyncR2SnapshotWrite(env, snapshotClaim(key), NOW, TOKEN_A);
|
|
await bucket.put(key, bytes("ciphertext-a"));
|
|
commitGenesis(databasePath, key, HASH_A, lease.writeToken);
|
|
|
|
const document = await syncResetDocument(
|
|
await resetRequest("sync-reset-000001", NOW + 1),
|
|
env,
|
|
authContext(),
|
|
NOW + 1,
|
|
);
|
|
|
|
assert.equal(document.deleted.snapshots, 1);
|
|
assert.equal(document.deleted.r2_objects, 1);
|
|
assert.deepEqual(query(databasePath, `
|
|
SELECT current_key_id, current_generation FROM sync_vault_accounts
|
|
WHERE user_id = '${USER_ID}'
|
|
`), [{ current_key_id: KEY_ID, current_generation: 1 }]);
|
|
assert.equal(query(databasePath, `
|
|
SELECT COUNT(*) AS count FROM sync_vault_envelopes WHERE user_id = '${USER_ID}'
|
|
`)[0]?.count, 1);
|
|
assert.equal(query(databasePath, `
|
|
SELECT COUNT(*) AS count FROM user_devices WHERE user_id = '${USER_ID}'
|
|
`)[0]?.count, 1);
|
|
assert.equal(query(databasePath, `
|
|
SELECT COUNT(*) AS count FROM sync_snapshots WHERE user_id = '${USER_ID}'
|
|
`)[0]?.count, 0);
|
|
assert.equal(candidateState(databasePath, key), "deleted");
|
|
assert.deepEqual(bucket.deletes, [key]);
|
|
});
|
|
});
|
|
|
|
it("keeps a fenced pending lease until a late R2 put can be collected", async () => {
|
|
await withDatabase(async (databasePath, _database, bucket, env) => {
|
|
const key = snapshotKey(HASH_A);
|
|
await claimSyncR2SnapshotWrite(env, snapshotClaim(key), NOW, TOKEN_A);
|
|
await syncResetDocument(
|
|
await resetRequest("sync-reset-000002", NOW + 1),
|
|
env,
|
|
authContext(),
|
|
NOW + 1,
|
|
);
|
|
|
|
assert.equal(candidateState(databasePath, key), "ready");
|
|
assert.equal(await collectSyncR2Garbage(env, NOW + 1, { userId: USER_ID }), 0);
|
|
await bucket.put(key, bytes("late-pending-ciphertext"));
|
|
assert.equal(await collectSyncR2Garbage(env, NOW + 600, { userId: USER_ID }), 1);
|
|
assert.equal(candidateState(databasePath, key), "deleted");
|
|
assert.equal(bucket.has(key), false);
|
|
assert.deepEqual(bucket.deletes, [key]);
|
|
assert.equal(query(databasePath, `
|
|
SELECT COUNT(*) AS count FROM sync_vault_accounts WHERE user_id = '${USER_ID}'
|
|
`)[0]?.count, 1);
|
|
});
|
|
});
|
|
|
|
it("rolls back reset when its authenticated authority changes before the batch", async () => {
|
|
for (const beforeBatchSql of [
|
|
"DELETE FROM better_auth_session WHERE id = 'session-01';",
|
|
`UPDATE user_devices SET revoked_at = ${NOW} WHERE device_id = '${DEVICE_ID}';`,
|
|
]) {
|
|
await withDatabase(async (databasePath, _database, bucket, env) => {
|
|
const key = snapshotKey(HASH_A);
|
|
const lease = await claimSyncR2SnapshotWrite(env, snapshotClaim(key), NOW, TOKEN_A);
|
|
await bucket.put(key, bytes("ciphertext-a"));
|
|
commitGenesis(databasePath, key, HASH_A, lease.writeToken);
|
|
const request = await resetRequest("sync-reset-authority-race", NOW + 1);
|
|
const racedEnv = {
|
|
...env,
|
|
ELY_DB: new SqliteD1Database(databasePath, beforeBatchSql),
|
|
} as Env;
|
|
|
|
await assert.rejects(
|
|
() => syncResetDocument(request, racedEnv, authContext(), NOW + 1),
|
|
/device_action_gate_failed/,
|
|
);
|
|
|
|
assert.deepEqual(query(databasePath, `SELECT
|
|
(SELECT COUNT(*) FROM sync_snapshots WHERE user_id = '${USER_ID}') AS snapshots,
|
|
(SELECT COUNT(*) FROM sync_snapshot_encryption WHERE user_id = '${USER_ID}') AS encryption,
|
|
(SELECT COUNT(*) FROM sync_snapshot_heads WHERE user_id = '${USER_ID}') AS heads,
|
|
(SELECT COUNT(*) FROM audit_events WHERE event_type = 'sync.reset') AS audits
|
|
`), [{ snapshots: 1, encryption: 1, heads: 1, audits: 0 }]);
|
|
assert.equal(candidateState(databasePath, key), "referenced");
|
|
assert.equal(bucket.has(key), true);
|
|
});
|
|
}
|
|
});
|
|
|
|
it("retries an idempotent R2 deletion after a crash before D1 finalization", async () => {
|
|
await withDatabase(async (databasePath, _database, bucket, env) => {
|
|
const key = snapshotKey(HASH_A);
|
|
const lease = await claimSyncR2SnapshotWrite(env, snapshotClaim(key), NOW, TOKEN_A);
|
|
await bucket.put(key, bytes("ciphertext-a"));
|
|
await abandonSyncR2Write(env, USER_ID, OWNER_HASH, key, lease.writeToken, NOW + 1);
|
|
bucket.crashAfterNextDelete = true;
|
|
|
|
await assert.rejects(() => collectSyncR2Garbage(env, NOW + 1), /simulated_delete_crash/);
|
|
assert.equal(candidateState(databasePath, key), "deleting");
|
|
assert.equal(bucket.has(key), false);
|
|
|
|
assert.equal(await collectSyncR2Garbage(env, NOW + 62), 1);
|
|
assert.equal(candidateState(databasePath, key), "deleted");
|
|
assert.deepEqual(bucket.deletes, [key, key]);
|
|
});
|
|
});
|
|
|
|
it("fences an upload that overlaps account deletion and clears the raw owner id", async () => {
|
|
await withDatabase(async (databasePath, database, bucket, env) => {
|
|
const key = snapshotKey(HASH_A);
|
|
const lease = await claimSyncR2SnapshotWrite(env, snapshotClaim(key), NOW, TOKEN_A);
|
|
await database.batch([
|
|
database.prepare(SYNC_R2_FENCE_USER_QUERY).bind(NOW + 1, NOW + 1, NOW + 1, USER_ID),
|
|
database.prepare(SYNC_R2_ANONYMIZE_USER_QUERY).bind(NOW + 1, USER_ID, OWNER_HASH),
|
|
]);
|
|
assert.equal(await collectSyncR2Garbage(env, NOW + 1, { ownerHash: OWNER_HASH }), 0);
|
|
await bucket.put(key, bytes("late-ciphertext"));
|
|
|
|
assert.throws(
|
|
() => commitGenesis(databasePath, key, HASH_A, lease.writeToken),
|
|
/sync_r2_write_fenced/,
|
|
);
|
|
assert.deepEqual(candidateOwner(databasePath, key), {
|
|
user_id: null,
|
|
owner_hash: OWNER_HASH,
|
|
state: "ready",
|
|
});
|
|
assert.equal(await collectSyncR2Garbage(
|
|
env,
|
|
lease.leaseExpiresAt,
|
|
{ ownerHash: OWNER_HASH },
|
|
), 1);
|
|
});
|
|
});
|
|
|
|
it("rejects a pre-put claim after reset removes the vault authority", async () => {
|
|
await withDatabase(async (_databasePath, database, _bucket, env) => {
|
|
await database.prepare("DELETE FROM sync_vault_accounts WHERE user_id = ?")
|
|
.bind(USER_ID)
|
|
.run();
|
|
|
|
await assert.rejects(
|
|
() => claimSyncR2SnapshotWrite(
|
|
env,
|
|
snapshotClaim(snapshotKey(HASH_A)),
|
|
NOW,
|
|
TOKEN_A,
|
|
),
|
|
/sync_r2_write_fenced/,
|
|
);
|
|
});
|
|
});
|
|
|
|
it("inventories historical snapshot and sync-payload orphans", async () => {
|
|
await withDatabase(async (databasePath, _database, bucket, env) => {
|
|
const snapshot = snapshotKey(HASH_A);
|
|
const payload = payloadKey(HASH_B);
|
|
await bucket.put(snapshot, bytes("snapshot-orphan"));
|
|
await bucket.put(payload, bytes("payload-orphan"));
|
|
|
|
assert.equal(await inventorySyncR2Objects(env, NOW, 100), 1);
|
|
assert.equal(await inventorySyncR2Objects(env, NOW + 1, 100), 1);
|
|
assert.equal(candidateState(databasePath, snapshot), "ready");
|
|
assert.equal(candidateState(databasePath, payload), "ready");
|
|
assert.equal(await collectSyncR2Garbage(env, NOW + 1), 2);
|
|
assert.deepEqual(bucket.deletes.sort(), [payload, snapshot].sort());
|
|
});
|
|
});
|
|
});
|
|
|
|
async function withDatabase(
|
|
assertions: (
|
|
databasePath: string,
|
|
database: SqliteD1Database,
|
|
bucket: TestBucket,
|
|
env: Env,
|
|
) => Promise<void>,
|
|
): Promise<void> {
|
|
const tempDir = mkdtempSync(join(tmpdir(), "ely-r2-gc-"));
|
|
try {
|
|
const databasePath = join(tempDir, "ely.db");
|
|
for (const fileName of readdirSync(MIGRATIONS_DIR).filter((name) => name.endsWith(".sql")).sort()) {
|
|
execute(databasePath, readFileSync(join(MIGRATIONS_DIR, fileName), "utf8"));
|
|
}
|
|
seedAuthority(databasePath);
|
|
const database = new SqliteD1Database(databasePath);
|
|
const bucket = new TestBucket();
|
|
const env = { ELY_DB: database, ELY_STORAGE: bucket } as unknown as Env;
|
|
await assertions(databasePath, database, bucket, env);
|
|
} finally {
|
|
rmSync(tempDir, { recursive: true, force: true });
|
|
}
|
|
}
|
|
|
|
function seedAuthority(databasePath: string): void {
|
|
execute(databasePath, `
|
|
INSERT INTO better_auth_user (
|
|
id, name, email, emailVerified, createdAt, updatedAt
|
|
) VALUES (
|
|
'${USER_ID}', 'User', 'user@example.com', 1,
|
|
'2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z'
|
|
);
|
|
INSERT INTO user_devices (
|
|
user_id, device_id, public_key, device_name, platform,
|
|
approval_status, created_at, approved_at, last_active_at, revoked_at, idempotency_key
|
|
) VALUES (
|
|
'${USER_ID}', '${DEVICE_ID}', '${PUBLIC_KEY}', 'Mac', 'macOS',
|
|
'approved', 1, 1, 1, NULL, 'device-register-0001'
|
|
);
|
|
INSERT INTO user_device_keys (
|
|
user_id, device_id, signing_public_key, wrapping_public_key,
|
|
key_protocol_version, created_at
|
|
) VALUES (
|
|
'${USER_ID}', '${DEVICE_ID}', '${PUBLIC_KEY}', '${"f".repeat(64)}', 2, 1
|
|
);
|
|
INSERT INTO better_auth_session (
|
|
id, expiresAt, token, createdAt, updatedAt, userId
|
|
) VALUES (
|
|
'session-01', '2099-01-01T00:00:00Z', 'session-token-01',
|
|
'2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z', '${USER_ID}'
|
|
);
|
|
INSERT INTO better_auth_session_device_context (
|
|
session_id, user_id, device_id, updated_at
|
|
) VALUES ('session-01', '${USER_ID}', '${DEVICE_ID}', 1);
|
|
INSERT INTO sync_vault_accounts (
|
|
user_id, current_key_id, current_generation, created_at, updated_at
|
|
) VALUES ('${USER_ID}', '${KEY_ID}', 1, 1, 1);
|
|
`);
|
|
}
|
|
|
|
function seedVaultEnvelope(databasePath: string): void {
|
|
execute(databasePath, `
|
|
INSERT INTO sync_vault_envelopes (
|
|
user_id, recipient_device_id, approver_device_id, key_id, generation,
|
|
envelope_version, suite, encapped_key, ciphertext, idempotency_key, created_at
|
|
) VALUES (
|
|
'${USER_ID}', '${DEVICE_ID}', '${DEVICE_ID}', '${KEY_ID}', 1, 1,
|
|
'HPKE-BASE-X25519-HKDF-SHA256-CHACHA20POLY1305',
|
|
'${"A".repeat(43)}', '${"B".repeat(64)}', 'vault-bootstrap-0001', 1
|
|
);
|
|
`);
|
|
}
|
|
|
|
function commitGenesis(databasePath: string, r2Key: string, payloadHash: string, token: string): void {
|
|
execute(databasePath, `
|
|
BEGIN IMMEDIATE;
|
|
INSERT INTO sync_snapshots (
|
|
user_id, snapshot_id, r2_key, payload_hash, schema_rev, logical_clock,
|
|
device_id, size_bytes, created_at, head_revision,
|
|
base_head_revision, base_snapshot_id, base_payload_hash
|
|
) VALUES (
|
|
'${USER_ID}', '${DEVICE_ID}', '${r2Key}', '${payloadHash}', 1, 1,
|
|
'${DEVICE_ID}', 12, ${NOW}, 1, NULL, NULL, NULL
|
|
);
|
|
INSERT INTO sync_snapshot_encryption (
|
|
user_id, snapshot_id, encryption_version, vault_generation, key_id, content_hash
|
|
) VALUES ('${USER_ID}', '${DEVICE_ID}', 2, 1, '${KEY_ID}', '${HASH_B}');
|
|
INSERT INTO sync_snapshot_heads (
|
|
user_id, head_revision, snapshot_id, payload_hash, updated_at
|
|
) VALUES ('${USER_ID}', 1, '${DEVICE_ID}', '${payloadHash}', ${NOW});
|
|
UPDATE sync_r2_gc_candidates
|
|
SET state = 'referenced', lease_expires_at = ${NOW},
|
|
updated_at = ${NOW}, referenced_at = ${NOW}
|
|
WHERE r2_key = '${r2Key}' AND user_id = '${USER_ID}'
|
|
AND state = 'pending' AND write_token = '${token}'
|
|
AND lease_expires_at >= ${NOW};
|
|
COMMIT;
|
|
`);
|
|
}
|
|
|
|
function snapshotClaim(r2Key: string) {
|
|
return {
|
|
userId: USER_ID,
|
|
deviceId: DEVICE_ID,
|
|
r2Key,
|
|
ownerHash: OWNER_HASH,
|
|
keyId: KEY_ID,
|
|
generation: 1,
|
|
headRevision: 1,
|
|
baseHead: null,
|
|
} as const;
|
|
}
|
|
|
|
function authContext() {
|
|
return {
|
|
userId: USER_ID,
|
|
deviceId: DEVICE_ID,
|
|
sessionId: "session-01",
|
|
tokenHash: HASH_B,
|
|
expiresAt: "2099-01-01T00:00:00Z",
|
|
createdAt: "2026-01-01T00:00:00Z",
|
|
} as const;
|
|
}
|
|
|
|
async function resetRequest(idempotencyKey: string, proofCreatedAt: number): Promise<Request> {
|
|
const confirmation = "delete-cloud-sync-data";
|
|
const actionProof = await signDeviceMessage(recentDeviceActionProofBytes({
|
|
action: "sync.reset",
|
|
userId: USER_ID,
|
|
sessionId: authContext().sessionId,
|
|
deviceId: DEVICE_ID,
|
|
confirmation,
|
|
idempotencyKey,
|
|
proofCreatedAt,
|
|
}));
|
|
return new Request("https://elydora.test/api/sync/reset", {
|
|
method: "POST",
|
|
headers: { "content-type": "application/json" },
|
|
body: JSON.stringify({
|
|
version: 2,
|
|
confirmation,
|
|
idempotency_key: idempotencyKey,
|
|
proof_created_at: proofCreatedAt,
|
|
action_proof: actionProof,
|
|
}),
|
|
});
|
|
}
|
|
|
|
function candidateState(databasePath: string, key: string): unknown {
|
|
return query(databasePath, `
|
|
SELECT state FROM sync_r2_gc_candidates WHERE r2_key = '${key}'
|
|
`)[0]?.state;
|
|
}
|
|
|
|
function candidateOwner(databasePath: string, key: string): Record<string, unknown> | undefined {
|
|
return query(databasePath, `
|
|
SELECT user_id, owner_hash, state FROM sync_r2_gc_candidates WHERE r2_key = '${key}'
|
|
`)[0];
|
|
}
|
|
|
|
function snapshotKey(hash: string): string {
|
|
return `sync-snapshots/us-east/${OWNER_HASH}/${DEVICE_ID}/${hash}.bin`;
|
|
}
|
|
|
|
function payloadKey(hash: string): string {
|
|
return `sync-payloads/us-east/${OWNER_HASH}/bookmarks/object-01/${hash}.bin`;
|
|
}
|
|
|
|
function bytes(value: string): ArrayBuffer {
|
|
return new TextEncoder().encode(value).buffer;
|
|
}
|
|
|
|
class TestBucket {
|
|
readonly deletes: string[] = [];
|
|
crashAfterNextDelete = false;
|
|
private readonly values = new Map<string, ArrayBuffer>();
|
|
|
|
get(key: string): Promise<ElyR2Object | null> {
|
|
const value = this.values.get(key);
|
|
return Promise.resolve(value === undefined ? null : object(value));
|
|
}
|
|
|
|
put(key: string, value: ArrayBuffer, _options?: ElyR2PutOptions): Promise<ElyR2Object> {
|
|
this.values.set(key, value);
|
|
return Promise.resolve(object(value));
|
|
}
|
|
|
|
async delete(key: string): Promise<void> {
|
|
this.deletes.push(key);
|
|
this.values.delete(key);
|
|
if (this.crashAfterNextDelete) {
|
|
this.crashAfterNextDelete = false;
|
|
throw new Error("simulated_delete_crash");
|
|
}
|
|
}
|
|
|
|
list(options: { prefix: string; cursor?: string; limit: number }) {
|
|
const keys = [...this.values.keys()].filter((key) => key.startsWith(options.prefix)).sort();
|
|
return Promise.resolve({
|
|
objects: keys.slice(0, options.limit).map((key) => ({ key })),
|
|
truncated: false as const,
|
|
});
|
|
}
|
|
|
|
has(key: string): boolean {
|
|
return this.values.has(key);
|
|
}
|
|
}
|
|
|
|
function object(value: ArrayBuffer): ElyR2Object {
|
|
return { arrayBuffer: () => Promise.resolve(value) };
|
|
}
|