Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 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 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 | 1x 1x 1x 1x 1x 1x 1x 1x 380x 380x 380x 380x 380x 380x 380x 1x 380x 380x 380x 380x 380x 1x 9449x 9449x 1x 768x 768x 768x 686x 686x 685x 685x 768x 1x 3891x 3891x 1x 10368x 10x 797x 7x 7x 10368x 1x 3x 3x 4x 4x 4x 2x 4x 3x 1x 3506x 3506x 3506x 3506x 3506x 1x 371x 371x 23x 23x 23x 348x 371x 371x 371x 371x 371x 1x 3828x 3828x 1x 2511x 320x 2511x 2511x 2511x 1x 168x 1x 1x 1x 168x 168x 1x 1075x 1075x 1075x 2x 2x 2x 2x 2x 2x 2x 1075x 1075x 1074x 1074x 1074x 1075x 1075x 1075x 1075x 1075x 1075x 1x 1074x 1074x 1053x 1053x 1053x 102x 1074x 14x 14x 3x 14x 1x 1x 7x 1074x 1x 529x 529x 1x 128x 128x 128x 128x 1048x 1048x 1048x 128x 128x 128x 128x 128x 128x 126x 126x 128x 2x 2x 128x 128x 128x 128x 128x 128x 128x 128x 128x 128x 128x 128x 1x 1434x 1434x 1434x 4x 1434x 1434x 1434x 1x 31x 31x 27x 31x 31x 31x 31x 1x 980x 980x 1x 966x 966x 1x 14x 14x 1x 618x 618x 618x 618x 618x 618x 618x 618x 618x 1x 5x 5x 5x 5x 1x | import { decodeJSON, encodeJSON, randomId } from './hash.js';
import { encodeRows, missingRows, parseRows, xorDigest } from './log.js';
import { concat, readRange } from './stream.js';
import {
DEFAULT_TEXT_EXTENSIONS,
HEADER_PROBE,
ZERO_DIGEST,
decodeVFSFile,
emptyFile,
encodeVFSFile,
headerOf,
normalizeFile,
parseHeader,
} from './vfs-file.js';
import type { Hash, LogRow, VFSAdapter, VFSEntry, VFSFile, VFSHeader } from './types.js';
/** Name of the control folder that lives inside every synced folder. */
export const CONTROL_DIR = '.vfs';
/**
* How big the active log segment is allowed to get before it rotates.
*
* This is literally "how much am I willing to re-upload on every sync": Drive
* has no append, so extending the log rewrites the whole active segment. 256 KB
* is the design's starting point. Rotation is also forced after a bulk import,
* which produces thousands of rows in one go.
*/
export const ROTATE_AT = 256 * 1024;
export interface VFSStoreOptions {
/** Segment size that triggers rotation. Defaults to {@link ROTATE_AT}. */
rotateAt?: number;
/** Injectable clock, for tests. */
now?: () => number;
}
/**
* The `.vfs/` control folder in v2 — two files on the normal path.
*
* ```
* .vfs/
* vfs.json # mutable: header + the mirror of the tree
* commits # append-only: the active segment of the log
* commits-<ts> # closed segments, immutable
* vfs-<ts>.json # cumulative snapshot taken when each segment closed
* base/<hash> # local only: previous text content, for diff3
* ```
*
* There is no object store: the working file *is* the content. What is left
* here is metadata, an append-only log of operations, and a local scratch area
* for three-way merges that never travels.
*/
export class VFSStore {
readonly adapter: VFSAdapter;
readonly root: string;
readonly rotateAt: number;
private file: VFSFile | null = null;
/** Active segment, in arrival order — the order it has on disk. */
private rows: LogRow[] | null = null;
/** Cumulative snapshot of the last rotation, decoded. */
private snapshot: VFSEntry[] | null = null;
/** Closed segments are immutable, so what came back once is kept for good. */
private readonly archives = new Map<number, LogRow[]>();
private readonly now: () => number;
constructor(adapter: VFSAdapter, root = CONTROL_DIR, options: VFSStoreOptions = {}) {
this.adapter = adapter;
this.root = root;
this.rotateAt = options.rotateAt ?? ROTATE_AT;
this.now = options.now ?? (() => Date.now());
}
path(...parts: string[]): string {
return [this.root, ...parts].join('/');
}
/**
* Reads a control file, or `null` when it is not there. Reads first and asks
* afterwards: a hit is one call instead of `stat` + `read`, and the `stat`
* only happens on the miss, where it still separates "absent" from "broken".
*/
private async readFile(path: string): Promise<Uint8Array | null> {
try {
return await this.adapter.read(path);
} catch (error) {
const stat = await this.adapter.stat(path);
if (stat && stat.kind === 'file') throw error;
return null;
}
}
// ------------------------------------------------------------- vfs.json
get filePath(): string {
return this.path('vfs.json');
}
async read(): Promise<VFSFile> {
if (this.file) return this.file;
const data = await this.readFile(this.filePath);
if (!data) throw new Error(`no vfs store in ${this.adapter.name}`);
this.file = decodeVFSFile(data);
return this.file;
}
/**
* Header only — `state`, `log`, `peers` — from a range read of the top of the
* file, growing the probe until `entries` comes into view.
*
* Not what `sync` uses: it drives both nodes and reconciling one needs its
* entries anyway. This is for inspecting a store you are *not* opening as a
* node — "does this folder hold a store, and which one?" — where `entries` can
* run to megabytes and none of it is needed.
*/
async header(): Promise<VFSHeader> {
if (this.file) return headerOf(this.file);
for (let probe = HEADER_PROBE; probe <= HEADER_PROBE * 16; probe *= 4) {
const prefix = await readRange(this.adapter, this.filePath, { end: probe });
const header = parseHeader(prefix);
if (header) return header;
if (prefix.byteLength < probe) break; // that was the whole file already
}
// Pathological header (hundreds of peers): fall back to reading it all.
return this.read();
}
/** Sorts, re-digests and writes. The only way `vfs.json` changes on disk. */
async write(file: VFSFile): Promise<VFSFile> {
const normalized = await normalizeFile(file);
this.file = normalized;
await this.adapter.write(this.filePath, encodeVFSFile(normalized));
return normalized;
}
/** Opens an existing store, or lays down a fresh one. */
async init(options: { peerId?: string } = {}): Promise<VFSFile> {
const data = await this.readFile(this.filePath);
if (data) {
this.file = decodeVFSFile(data);
return this.file;
}
return this.write(
emptyFile(options.peerId ?? randomId(), this.now(), [
...DEFAULT_TEXT_EXTENSIONS,
]),
);
}
// ------------------------------------------------------------ commit log
segmentPath(segment?: number): string {
return segment === undefined ? this.path('commits') : this.path(`commits-${segment}`);
}
/** The whole active segment, in arrival order. */
async logRows(): Promise<LogRow[]> {
if (this.rows) return this.rows;
const data = await this.readFile(this.segmentPath());
this.rows = data ? parseRows(data) : [];
return this.rows;
}
/**
* Rows of the active segment from a byte offset — one range read instead of
* the whole file. Returning a superset is always safe (union deduplicates by
* `op`), so a cached segment answers from memory.
*/
async rowsSince(offset: number): Promise<LogRow[]> {
if (this.rows || offset <= 0) return this.logRows();
const data = await readRange(this.adapter, this.segmentPath(), { start: offset }).catch(
() => null,
);
return data ? parseRows(data) : [];
}
/**
* Adds the rows the segment does not already hold and refreshes `log.*`.
*
* There is only one writer per store, and a shared file means two concurrent
* appenders lose rows, so the mitigation is explicit: compare the file's size
* against what the header claims and, if it grew underneath us, fold the tail
* in before writing. Where the backend offers a conditional write, that
* check-then-act becomes atomic rather than merely likely.
*/
async append(rows: LogRow[], file?: VFSFile): Promise<VFSFile> {
const target = file ?? (await this.read());
const existing = await this.logRows();
const onDisk = (await this.adapter.stat(this.segmentPath()))?.size ?? 0;
if (onDisk > target.log.size) {
const extra = parseRows(
await readRange(this.adapter, this.segmentPath(), { start: target.log.size }).catch(
() => new Uint8Array(),
),
);
for (const row of missingRows(existing, extra)) existing.push(row);
}
const fresh = missingRows(existing, rows);
if (fresh.length > 0) {
await this.appendBytes(this.segmentPath(), encodeRows(fresh));
for (const row of fresh) existing.push(row);
}
this.rows = existing;
target.log.rows = existing.length;
target.log.digest = xorDigest(existing);
target.log.size = (await this.adapter.stat(this.segmentPath()))?.size ?? target.log.size;
return target;
}
/** Native append where the backend has one, read-concat-write where it does not. */
private async appendBytes(path: string, data: Uint8Array): Promise<void> {
if (data.byteLength === 0) return;
if (this.adapter.append) {
await this.adapter.append(path, data);
return;
}
const held = (await this.readFile(path)) ?? new Uint8Array();
if (this.adapter.writeIf && this.adapter.tag) {
const tag = held.byteLength > 0 ? await this.adapter.tag(path) : null;
if ((await this.adapter.writeIf(path, concat([held, data]), tag)) !== null) return;
// Lost the race: whoever won is on disk now, so re-read and fold in on top.
const again = (await this.readFile(path)) ?? new Uint8Array();
await this.adapter.write(path, concat([again, data]));
return;
}
await this.adapter.write(path, concat([held, data]));
}
// -------------------------------------------------------------- rotation
/** True when the active segment has outgrown the threshold. */
async shouldRotate(file?: VFSFile): Promise<boolean> {
return (file ?? (await this.read())).log.size >= this.rotateAt;
}
/**
* Closes the active segment and takes a cumulative snapshot.
*
* The order is the whole point, and it is what makes tombstone pruning safe:
* **rotate and photograph first, prune `vfs.json` afterwards**. The snapshot
* is *previous snapshot ∪ current entries*, tombstones included — not "the
* tree at this moment" but "the last known state of every uuid that has ever
* existed". A snapshot that was merely the current tree would drop a delete
* whose tombstone had been pruned two segments ago, and the file would
* resurrect on the next peer that still had it.
*/
async rotate(file?: VFSFile): Promise<VFSFile> {
const target = file ?? (await this.read());
const closing = target.log.segment;
const rows = await this.logRows();
const merged = new Map<string, VFSEntry>();
for (const entry of await this.readSnapshot(target)) merged.set(entry.uuid, entry);
for (const entry of target.entries) {
const held = merged.get(entry.uuid);
if (!held || entry.updated >= held.updated) merged.set(entry.uuid, entry);
}
const entries = [...merged.values()];
const stamp = Math.max(this.now(), closing + 1);
const snapshot = `vfs-${stamp}.json`;
await this.adapter.write(this.path(snapshot), encodeJSON({ entries }));
const archives = [...(target.log.archives ?? [])];
if (rows.length > 0) {
await this.adapter.rename(this.segmentPath(), this.segmentPath(closing));
archives.push(closing);
} else {
await this.adapter.delete(this.segmentPath()).catch(() => undefined);
}
this.snapshot = entries;
this.rows = [];
target.log = {
segment: stamp,
digest: ZERO_DIGEST,
rows: 0,
size: 0,
snapshot,
...(archives.length > 0 ? { archives } : {}),
};
return target;
}
/** Entries of the cumulative snapshot named in the header, or `[]`. */
async readSnapshot(file?: VFSFile | VFSHeader): Promise<VFSEntry[]> {
if (this.snapshot) return this.snapshot;
const name = (file ?? (await this.read())).log.snapshot;
if (!name) return (this.snapshot = []);
const data = await this.readFile(this.path(name));
this.snapshot = data ? (decodeJSON<{ entries: VFSEntry[] }>(data).entries ?? []) : [];
return this.snapshot;
}
/**
* Rows of a closed segment. Immutable, so it is cached for good — and it is
* only ever read to *avoid* a conflict copy, never to decide state, so a
* missing archive costs a spurious copy and nothing else.
*/
async readArchive(segment: number): Promise<LogRow[]> {
const held = this.archives.get(segment);
if (held) return held;
const data = await this.readFile(this.segmentPath(segment));
const rows = data ? parseRows(data) : [];
this.archives.set(segment, rows);
return rows;
}
// ------------------------------------------------------------------ base
basePath(hash: Hash): string {
return this.path('base', hash);
}
/** Keeps the text content that is about to be overwritten, for a later diff3. */
async putBase(hash: Hash, data: Uint8Array): Promise<void> {
await this.adapter.write(this.basePath(hash), data);
}
async getBase(hash: Hash): Promise<Uint8Array | null> {
return this.readFile(this.basePath(hash));
}
/**
* Drops base copies nothing can still need.
*
* `keep` is derived from a signal that already exists — content still
* referenced back to the oldest `peers[*].lastSync` — so retention prunes
* itself at the pace of syncing. `limit` is the hard stop for the risk the
* design calls out: a peer configured once and forgotten would otherwise hold
* the window open forever. Over-pruning only degrades a text merge to LWW
* plus a copy.
*/
async pruneBase(keep: Set<Hash>, limit = 256): Promise<void> {
const listing = (await this.adapter.list(this.path('base')).catch(() => [])).filter(
(entry) => entry.kind === 'file',
);
const doomed = listing.filter((entry) => !keep.has(entry.name));
// Over the hard cap even after dropping the unreferenced ones: give up the
// oldest of what is left. Losing a base only costs a merge that falls back
// to LWW plus a copy, so an unbounded folder is the worse outcome.
const held = listing.filter((entry) => keep.has(entry.name));
const overflow = held.length - limit;
if (overflow > 0) {
doomed.push(
...[...held].sort((x, y) => (x.stat?.mtime ?? 0) - (y.stat?.mtime ?? 0)).slice(0, overflow),
);
}
for (const entry of doomed) await this.adapter.delete(entry.path).catch(() => undefined);
}
/** Drops memoised state so the next read hits the backend again. */
invalidate(): void {
this.file = null;
this.rows = null;
this.snapshot = null;
}
}
|