系統設計

1type Level = "ru" | "rc" | "rr" | "serializable";
2type Version = { value: number; ts: number };
3 
4class Store {
5  versions = new Map<string, Version[]>();
6  dirty = new Map<string, Array<{ value: number; tx: number }>>(); // uncommitted writes, latest last
7  wal: string[] = [];
8  clock = 0;
9  nextTx = 1;
10 
11  committed(key: string, at = this.clock): number | undefined {
12    const list = this.versions.get(key) ?? [];
13    for (let i = list.length - 1; i >= 0; i--) if (list[i].ts <= at) return list[i].value;
14    return undefined;
15  }
16 
17  changedSince(key: string, ts: number): boolean {
18    return (this.versions.get(key) ?? []).some((v) => v.ts > ts);
19  }
20 
21  keys(): string[] {
22    return [...new Set([...this.versions.keys(), ...this.dirty.keys()])].sort();
23  }
24}
25 
26class Tx {
27  snapshot: number;
28  id: number;
29  writes = new Map<string, number>();
30  reads = new Set<string>();
31  scans = new Set<string>();
32 
33  constructor(private db: Store, private level: Level) {
34    this.snapshot = db.clock;
35    this.id = db.nextTx++;
36  }
37 
38  read(key: string): number | undefined {
39    if (this.writes.has(key)) return this.writes.get(key);
40    this.reads.add(key);
41    if (this.level === "ru" && this.db.dirty.has(key)) {
42      return this.db.dirty.get(key)!.at(-1)!.value;
43    }
44    const fromSnapshot = this.level === "rr" || this.level === "serializable";
45    return this.db.committed(key, fromSnapshot ? this.snapshot : this.db.clock);
46  }
47 
48  scan(prefix: string, min: number): number {
49    this.scans.add(prefix);
50    let count = 0;
51    for (const key of this.db.keys()) {
52      if (key.startsWith(prefix) && (this.read(key) ?? -Infinity) >= min) count++;
53    }
54    return count;
55  }
56 
57  write(key: string, value: number): void {
58    this.writes.set(key, value);
59    // Visibility depends on the reader's level, not the writer's.
60    const pending = (this.db.dirty.get(key) ?? []).filter(e => e.tx !== this.id);
61    pending.push({ value, tx: this.id });
62    this.db.dirty.set(key, pending);
63  }
64 
65  commit(): boolean {
66    if (this.level === "rr" || this.level === "serializable") {
67      for (const key of this.writes.keys()) {
68        if (this.db.changedSince(key, this.snapshot)) return this.abort();
69      }
70    }
71    if (this.level === "serializable") {
72      for (const key of this.reads) {
73        if (this.db.changedSince(key, this.snapshot)) return this.abort();
74      }
75      for (const prefix of this.scans) for (const key of this.db.keys()) {
76        if (key.startsWith(prefix) && this.db.changedSince(key, this.snapshot)) return this.abort();
77      }
78    }
79    this.db.wal.push(JSON.stringify([...this.writes]));
80    // A real engine fsyncs the log here, before acknowledging.
81    const ts = ++this.db.clock;
82    for (const [key, value] of this.writes) {
83      const list = this.db.versions.get(key) ?? [];
84      list.push({ value, ts });
85      this.db.versions.set(key, list);
86    }
87    this.clearDirty();
88    return true;
89  }
90 
91  abort(): false {
92    this.writes.clear();
93    this.clearDirty();
94    return false;
95  }
96 
97  private clearDirty(): void {
98    for (const [key, entries] of this.db.dirty) {
99      const pending = entries.filter(e => e.tx !== this.id);
100      if (pending.length) this.db.dirty.set(key, pending);
101      else this.db.dirty.delete(key);
102    }
103  }
104}