File size: 3,800 Bytes
4e23b01 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 | // src/cluster/coordinator.ts
//
// Cross-shard write coordination. Ops are grouped by shard; groups are
// executed in ascending shard-id order (the fixed lock-acquisition order that
// avoids deadlocks between processes). Within one shard the group commits as
// a single MiniDb batch (one WAL frame, all-or-nothing); across shards the
// semantics depend on the configured CrossShardMode.
import type { BatchInputOp, MiniDb } from '../index.js';
import type { CrossShardMode } from './types.js';
import type { Router } from './router.js';
export class Coordinator<V = unknown> {
constructor(
private readonly router: Router,
private readonly runOnShard: <T>(shardId: number, fn: (db: MiniDb<V>) => T | Promise<T>) => Promise<T>,
private readonly mode: CrossShardMode,
) {}
private checkMode(shardIds: number[]): void {
if (shardIds.length <= 1) return;
if (this.mode === 'none') {
throw new Error(`operation spans ${shardIds.length} shards but crossShard mode is 'none'`);
}
if (this.mode === '2pc') {
throw new Error("crossShard mode '2pc' is reserved but not implemented yet");
}
}
/** Group items by their key's shard, shards in ascending id order. */
private group<T>(items: readonly T[], keyOf: (item: T) => string): [number, T[]][] {
const byShard = new Map<number, T[]>();
for (const item of items) {
const id = this.router.shardFor(keyOf(item));
const arr = byShard.get(id);
if (arr) arr.push(item);
else byShard.set(id, [item]);
}
return [...byShard.entries()].sort((a, b) => a[0] - b[0]);
}
private async runGroups<T>(
groups: [number, T[]][],
run: (shardId: number, items: T[]) => Promise<void>,
opName: string,
): Promise<void> {
this.checkMode(groups.map(([id]) => id));
const errors: unknown[] = [];
for (const [id, items] of groups) {
try {
await run(id, items);
} catch (e) {
// best-effort: earlier groups may be committed already; report, don't hide.
errors.push(e);
}
}
if (errors.length > 0) {
throw new AggregateError(errors, `${opName} failed on ${errors.length}/${groups.length} shard(s); partial writes possible`);
}
}
/** Multi-shard mset. Atomic within each shard (single WAL batch frame). */
async mset(entries: readonly (readonly [string, V])[]): Promise<void> {
const groups = this.group(entries, ([key]) => key);
await this.runGroups(
groups,
(id, items) =>
this.runOnShard(id, (db) => db.batch(items.map(([key, value]) => ({ op: 'set' as const, key, value })))),
'mset',
);
}
/** Multi-shard delete. Returns the number of keys that actually existed. */
async mdel(keys: readonly string[]): Promise<number> {
const groups = this.group(keys, (k) => k);
this.checkMode(groups.map(([id]) => id));
let removed = 0;
const errors: unknown[] = [];
for (const [id, ks] of groups) {
try {
removed += await this.runOnShard(id, async (db) => {
const existing = ks.filter((k) => db.has(k));
if (existing.length > 0) await db.batch(existing.map((key) => ({ op: 'del' as const, key })));
return existing.length;
});
} catch (e) {
errors.push(e);
}
}
if (errors.length > 0) {
throw new AggregateError(errors, `mdel failed on ${errors.length}/${groups.length} shard(s); partial writes possible`);
}
return removed;
}
/** Multi-shard atomic-per-shard batch (set/del ops with optional ttl/dt). */
async batch(ops: readonly BatchInputOp<V>[]): Promise<void> {
const groups = this.group(ops, (op) => op.key);
await this.runGroups(groups, (id, items) => this.runOnShard(id, (db) => db.batch(items)), 'batch');
}
}
|