Simplify DB
This commit is contained in:
parent
9244e5d220
commit
3177691e13
2 changed files with 98 additions and 316 deletions
|
|
@ -4,7 +4,7 @@ import { Database, createEmptyDatabase } from "./database";
|
||||||
|
|
||||||
test("current state keeps absent optional fields absent across save and reload", async () => {
|
test("current state keeps absent optional fields absent across save and reload", async () => {
|
||||||
const initial = createEmptyDatabase("vault");
|
const initial = createEmptyDatabase("vault");
|
||||||
initial.local = { local: "local.md", remote: "remote.md" };
|
initial.actualFileManifest = { local: "local.md", remote: "remote.md" };
|
||||||
initial.documents = {
|
initial.documents = {
|
||||||
local: { observedHash: "local" },
|
local: { observedHash: "local" },
|
||||||
remote: { remote: { vaultUpdateId: 1, contentSize: 3 } },
|
remote: { remote: { vaultUpdateId: 1, contentSize: 3 } },
|
||||||
|
|
@ -29,7 +29,7 @@ test("current state keeps absent optional fields absent across save and reload",
|
||||||
|
|
||||||
test("current reset state is restored after an uncertain save", async () => {
|
test("current reset state is restored after an uncertain save", async () => {
|
||||||
const saved = createEmptyDatabase("vault");
|
const saved = createEmptyDatabase("vault");
|
||||||
saved.local = { note: "note.md" };
|
saved.actualFileManifest = { note: "note.md" };
|
||||||
saved.documents.note = {
|
saved.documents.note = {
|
||||||
base: { vaultUpdateId: null, hash: "clean" },
|
base: { vaultUpdateId: null, hash: "clean" },
|
||||||
observedHash: "clean"
|
observedHash: "clean"
|
||||||
|
|
@ -43,6 +43,6 @@ test("current reset state is restored after an uncertain save", async () => {
|
||||||
async () => structuredClone(saved)
|
async () => structuredClone(saved)
|
||||||
);
|
);
|
||||||
await assert.rejects(db.save(), /interrupted/);
|
await assert.rejects(db.save(), /interrupted/);
|
||||||
await db.recoverPersistence();
|
await db.reloadFromSave();
|
||||||
assert.deepEqual(db.state, saved);
|
assert.deepEqual(db.state, saved);
|
||||||
});
|
});
|
||||||
|
|
|
||||||
|
|
@ -1,374 +1,156 @@
|
||||||
import type { Logger } from "../tracing/logger";
|
import type { PendingRequest } from "./requests";
|
||||||
import { EMPTY_HASH } from "../utils/hash";
|
import type { FileManifest } from "../services/types/FileManifest";
|
||||||
import { CoveredValues } from "../utils/data-structures/min-covered";
|
import type { FileManifestEntries } from "../types/file-manifest-entries";
|
||||||
import { awaitAll } from "../utils/await-all";
|
|
||||||
import { removeFromArray } from "../utils/remove-from-array";
|
|
||||||
|
|
||||||
export type VaultUpdateId = number;
|
export type VaultUpdateId = number;
|
||||||
export type DocumentId = string;
|
export type DocumentId = string;
|
||||||
export type RelativePath = string;
|
export type RelativePath = string;
|
||||||
|
|
||||||
export interface DocumentMetadata {
|
export interface ContentHead {
|
||||||
parentVersionId: VaultUpdateId;
|
vaultUpdateId: VaultUpdateId;
|
||||||
hash: string;
|
contentSize: number;
|
||||||
remoteRelativePath?: RelativePath;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
export interface StoredDocumentMetadata {
|
export interface DocumentState {
|
||||||
relativePath: RelativePath;
|
// A null version retains only the clean hash after a server history reset.
|
||||||
documentId: DocumentId;
|
base?: { vaultUpdateId: VaultUpdateId | null; hash: string };
|
||||||
parentVersionId: VaultUpdateId;
|
|
||||||
remoteRelativePath?: RelativePath;
|
// Last observed disk hash; an empty string means a notification awaits scanning.
|
||||||
hash: string;
|
observedHash?: string;
|
||||||
|
|
||||||
|
// A remote version still to be incorporated, including excluded downloads.
|
||||||
|
remote?: ContentHead;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export type { PendingRequest } from "./requests";
|
||||||
|
|
||||||
export interface StoredDatabase {
|
export interface StoredDatabase {
|
||||||
documents: StoredDocumentMetadata[];
|
readonly vaultKey: string;
|
||||||
lastSeenUpdateId: VaultUpdateId | undefined;
|
|
||||||
hasInitialSyncCompleted: boolean;
|
// Last incorporated server manifest; the base for the next three-way path merge.
|
||||||
|
parentFileManifest: FileManifest;
|
||||||
|
|
||||||
|
// Current local document IDs and paths, including files awaiting upload or download.
|
||||||
|
actualFileManifest: FileManifestEntries;
|
||||||
|
|
||||||
|
// Per-document merge bases and observations; content bytes are not stored here.
|
||||||
|
documents: Record<DocumentId, DocumentState>;
|
||||||
|
|
||||||
|
// Absent until a vault snapshot is applied; also marks a durable history reset.
|
||||||
|
lastSeenUpdateId?: VaultUpdateId;
|
||||||
|
|
||||||
|
// Exact request to resolve before observing a newer remote base.
|
||||||
|
pending?: PendingRequest;
|
||||||
|
|
||||||
|
// Commits consumption of the saved notification queue with the identity map.
|
||||||
|
lastAppliedLocalChangeId?: string;
|
||||||
|
|
||||||
|
// Server path at the start of exclusion, retained as the path-merge base.
|
||||||
|
// null means the document was absent from the server manifest at that time.
|
||||||
|
excluded?: Record<DocumentId, RelativePath | null>;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
export function createEmptyDatabase(vaultKey: string): StoredDatabase {
|
||||||
* Represents a document in the database.
|
return {
|
||||||
*
|
vaultKey,
|
||||||
* It is mutable and its content should always represent the latest
|
parentFileManifest: { fileManifestId: 0, entries: {} },
|
||||||
* state of the document on disk based on the update events we have seen.
|
actualFileManifest: {},
|
||||||
*/
|
documents: {}
|
||||||
export interface DocumentRecord {
|
};
|
||||||
relativePath: RelativePath;
|
}
|
||||||
documentId: DocumentId;
|
|
||||||
metadata: DocumentMetadata | undefined;
|
function canBindToAnotherVault(state: StoredDatabase): boolean {
|
||||||
isDeleted: boolean;
|
return (
|
||||||
updates: Promise<unknown>[];
|
state.lastSeenUpdateId === undefined &&
|
||||||
parallelVersion: number;
|
!state.pending &&
|
||||||
|
Object.keys(state.actualFileManifest).length === 0
|
||||||
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
export class Database {
|
export class Database {
|
||||||
private documents: DocumentRecord[];
|
public state: StoredDatabase;
|
||||||
private lastSeenUpdateIds: CoveredValues;
|
private needsReload = false;
|
||||||
private hasInitialSyncCompleted: boolean;
|
|
||||||
|
|
||||||
public constructor(
|
public constructor(
|
||||||
private readonly logger: Logger,
|
initial: StoredDatabase | undefined,
|
||||||
initialState: Partial<StoredDatabase> | undefined,
|
private readonly saveData: (data: StoredDatabase) => Promise<void>,
|
||||||
private readonly saveData: (data: StoredDatabase) => Promise<void>
|
vaultKey: string,
|
||||||
|
private readonly loadData: () => Promise<StoredDatabase | undefined>
|
||||||
) {
|
) {
|
||||||
initialState ??= {};
|
const state =
|
||||||
|
initial === undefined
|
||||||
|
? createEmptyDatabase(vaultKey)
|
||||||
|
: structuredClone(initial);
|
||||||
|
|
||||||
this.documents =
|
if (state.vaultKey !== vaultKey && !canBindToAnotherVault(state)) {
|
||||||
initialState.documents?.map(
|
throw new Error(
|
||||||
({ relativePath, documentId, ...metadata }) => ({
|
"Incompatible vault identity. Preserve it for recovery and initialize a separate state store."
|
||||||
relativePath,
|
);
|
||||||
documentId,
|
}
|
||||||
metadata,
|
|
||||||
isDeleted: false,
|
|
||||||
updates: [],
|
|
||||||
parallelVersion: 0
|
|
||||||
})
|
|
||||||
) ?? [];
|
|
||||||
|
|
||||||
this.ensureConsistency();
|
this.state = { ...state, vaultKey };
|
||||||
this.logger.debug(`Loaded ${this.documents.length} documents`);
|
|
||||||
|
|
||||||
const { lastSeenUpdateId } = initialState;
|
|
||||||
this.logger.debug(`Loaded last seen update id: ${lastSeenUpdateId}`);
|
|
||||||
this.lastSeenUpdateIds = new CoveredValues(
|
|
||||||
Math.max(0, lastSeenUpdateId ?? 0) // the first updateId will be 1 which is the first integer after -1
|
|
||||||
);
|
|
||||||
|
|
||||||
this.documents.forEach((doc) => {
|
|
||||||
this.lastSeenUpdateIds.add(doc.metadata?.parentVersionId);
|
|
||||||
});
|
|
||||||
|
|
||||||
this.hasInitialSyncCompleted =
|
|
||||||
initialState.hasInitialSyncCompleted ?? false;
|
|
||||||
this.logger.debug(
|
|
||||||
`Loaded hasInitialSyncCompleted: ${this.hasInitialSyncCompleted}`
|
|
||||||
);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public get length(): number {
|
public get length(): number {
|
||||||
return this.documents.length;
|
return Object.keys(this.state.actualFileManifest).length;
|
||||||
}
|
}
|
||||||
|
|
||||||
public get resolvedDocuments(): DocumentRecord[] {
|
// Bind an empty database to the configured vault.
|
||||||
const paths = new Map<string, DocumentRecord[]>();
|
public async bindVault(vaultKey: string): Promise<void> {
|
||||||
this.documents
|
if (this.state.vaultKey === vaultKey) {
|
||||||
// eslint-disable-next-line no-restricted-syntax -- Type narrowing, not removing a specific item
|
|
||||||
.filter(({ metadata }) => metadata !== undefined)
|
|
||||||
.forEach((record) =>
|
|
||||||
paths.set(record.relativePath, [
|
|
||||||
record,
|
|
||||||
...(paths.get(record.relativePath) ?? [])
|
|
||||||
])
|
|
||||||
);
|
|
||||||
|
|
||||||
return Array.from(paths.values()).map((records) => {
|
|
||||||
records.sort(
|
|
||||||
(a, b) => b.parallelVersion - a.parallelVersion // descending
|
|
||||||
);
|
|
||||||
|
|
||||||
if (
|
|
||||||
records.length > 1 &&
|
|
||||||
records.some((current, i) =>
|
|
||||||
i === 0
|
|
||||||
? false
|
|
||||||
: records[i - 1].parallelVersion ===
|
|
||||||
current.parallelVersion
|
|
||||||
)
|
|
||||||
) {
|
|
||||||
throw new Error(
|
|
||||||
`Multiple documents with the same parallel version and path at ${records[0].relativePath}`
|
|
||||||
);
|
|
||||||
}
|
|
||||||
return records[0];
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
public updateDocumentMetadata(
|
|
||||||
metadata: {
|
|
||||||
parentVersionId: VaultUpdateId;
|
|
||||||
hash: string;
|
|
||||||
remoteRelativePath: RelativePath;
|
|
||||||
},
|
|
||||||
toUpdate: DocumentRecord
|
|
||||||
): void {
|
|
||||||
if (!this.documents.includes(toUpdate)) {
|
|
||||||
throw new Error("Document not found in database");
|
|
||||||
}
|
|
||||||
|
|
||||||
toUpdate.metadata = metadata;
|
|
||||||
|
|
||||||
this.saveInTheBackground();
|
|
||||||
}
|
|
||||||
|
|
||||||
public removeDocumentPromise(promise: Promise<unknown>): void {
|
|
||||||
const entry = this.documents.find(({ updates }) =>
|
|
||||||
updates.includes(promise)
|
|
||||||
);
|
|
||||||
|
|
||||||
if (entry === undefined) {
|
|
||||||
// This method should be idempotent and tolerant of
|
|
||||||
// stragglers calling it after the databse has been reset.
|
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
removeFromArray(entry.updates, promise);
|
if (!canBindToAnotherVault(this.state)) {
|
||||||
// No need to save as Promises don't get serialized
|
|
||||||
}
|
|
||||||
|
|
||||||
public removeDocument(find: DocumentRecord): void {
|
|
||||||
removeFromArray(this.documents, find);
|
|
||||||
this.saveInTheBackground();
|
|
||||||
}
|
|
||||||
|
|
||||||
public getLatestDocumentByRelativePath(
|
|
||||||
find: RelativePath
|
|
||||||
): DocumentRecord | undefined {
|
|
||||||
const candidates = this.documents.filter(
|
|
||||||
({ relativePath }) => relativePath === find
|
|
||||||
);
|
|
||||||
candidates.sort((a, b) => b.parallelVersion - a.parallelVersion); // descending
|
|
||||||
return candidates[0];
|
|
||||||
}
|
|
||||||
|
|
||||||
public async getResolvedDocumentByRelativePath(
|
|
||||||
relativePath: RelativePath,
|
|
||||||
promise: Promise<unknown>
|
|
||||||
): Promise<DocumentRecord> {
|
|
||||||
const entry = this.getLatestDocumentByRelativePath(relativePath);
|
|
||||||
|
|
||||||
if (entry === undefined) {
|
|
||||||
throw new Error(
|
throw new Error(
|
||||||
`Document not found by relative path: ${relativePath}, ${JSON.stringify(
|
"Incompatible vault identity. Preserve it for recovery and initialize a separate state store."
|
||||||
this.documents,
|
|
||||||
null,
|
|
||||||
2
|
|
||||||
)}`
|
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
const currentPromises = entry.updates;
|
await this.commit({ ...this.state, vaultKey });
|
||||||
entry.updates = [...currentPromises, promise];
|
|
||||||
await awaitAll(currentPromises);
|
|
||||||
|
|
||||||
return entry;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public createNewPendingDocument(
|
public async commit(next: StoredDatabase): Promise<void> {
|
||||||
documentId: DocumentId,
|
const snapshot = structuredClone(next);
|
||||||
relativePath: RelativePath,
|
|
||||||
promise: Promise<unknown>
|
if (this.needsReload) {
|
||||||
): DocumentRecord {
|
throw new Error(
|
||||||
this.logger.debug(
|
"Reload saved state before retrying an uncertain save"
|
||||||
`Creating new pending document: ${relativePath} (${documentId})`
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
this.needsReload = true;
|
||||||
|
await this.saveData(snapshot);
|
||||||
|
this.state = snapshot;
|
||||||
|
this.needsReload = false;
|
||||||
|
}
|
||||||
|
|
||||||
|
public findDocumentId(path: RelativePath): DocumentId | undefined {
|
||||||
|
return Object.keys(this.state.actualFileManifest).find(
|
||||||
|
(id) => this.state.actualFileManifest[id] === path
|
||||||
);
|
);
|
||||||
const previousEntry =
|
|
||||||
this.getLatestDocumentByRelativePath(relativePath);
|
|
||||||
|
|
||||||
const entry = {
|
|
||||||
relativePath,
|
|
||||||
documentId,
|
|
||||||
metadata: undefined,
|
|
||||||
isDeleted: false,
|
|
||||||
updates: [promise],
|
|
||||||
parallelVersion:
|
|
||||||
previousEntry?.parallelVersion === undefined
|
|
||||||
? 0
|
|
||||||
: previousEntry.parallelVersion + 1
|
|
||||||
};
|
|
||||||
|
|
||||||
this.documents.push(entry);
|
|
||||||
this.saveInTheBackground();
|
|
||||||
|
|
||||||
return entry;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public createNewEmptyDocument(
|
public async reloadFromSave(): Promise<void> {
|
||||||
documentId: DocumentId,
|
if (!this.needsReload) {
|
||||||
parentVersionId: VaultUpdateId,
|
|
||||||
relativePath: RelativePath
|
|
||||||
): DocumentRecord {
|
|
||||||
const entry = {
|
|
||||||
relativePath,
|
|
||||||
documentId,
|
|
||||||
metadata: {
|
|
||||||
parentVersionId,
|
|
||||||
hash: EMPTY_HASH,
|
|
||||||
remoteRelativePath: relativePath
|
|
||||||
},
|
|
||||||
isDeleted: false,
|
|
||||||
updates: [],
|
|
||||||
parallelVersion: 0
|
|
||||||
};
|
|
||||||
|
|
||||||
this.documents.push(entry);
|
|
||||||
this.saveInTheBackground();
|
|
||||||
|
|
||||||
return entry;
|
|
||||||
}
|
|
||||||
|
|
||||||
public getDocumentByDocumentId(
|
|
||||||
find: DocumentId
|
|
||||||
): DocumentRecord | undefined {
|
|
||||||
return this.documents.find(({ documentId }) => documentId === find);
|
|
||||||
}
|
|
||||||
|
|
||||||
public move(
|
|
||||||
oldRelativePath: RelativePath,
|
|
||||||
newRelativePath: RelativePath
|
|
||||||
): void {
|
|
||||||
const oldDocument =
|
|
||||||
this.getLatestDocumentByRelativePath(oldRelativePath);
|
|
||||||
|
|
||||||
if (oldDocument === undefined) {
|
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
const newDocument =
|
const saved = await this.loadData();
|
||||||
this.getLatestDocumentByRelativePath(newRelativePath);
|
if (saved) {
|
||||||
if (newDocument?.isDeleted === false) {
|
this.state = structuredClone(saved);
|
||||||
throw new Error(
|
|
||||||
`Document already exists at new location: ${newRelativePath}`
|
|
||||||
);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
oldDocument.relativePath = newRelativePath;
|
this.needsReload = false;
|
||||||
// We're in a strange state where the target of the move has just got deleted,
|
|
||||||
// however, its metadata might already have a bunch of updates queued up for
|
|
||||||
// the document at the new location. We need to keep these updates.
|
|
||||||
oldDocument.parallelVersion =
|
|
||||||
newDocument !== undefined ? newDocument.parallelVersion + 1 : 0;
|
|
||||||
|
|
||||||
this.saveInTheBackground();
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public delete(relativePath: RelativePath): void {
|
public cloneState(): StoredDatabase {
|
||||||
const candidate = this.getLatestDocumentByRelativePath(relativePath);
|
return structuredClone(this.state);
|
||||||
if (candidate === undefined) {
|
|
||||||
throw new Error(
|
|
||||||
`Document not found by relative path: ${relativePath}`
|
|
||||||
);
|
|
||||||
}
|
|
||||||
candidate.isDeleted = true;
|
|
||||||
}
|
|
||||||
|
|
||||||
public getHasInitialSyncCompleted(): boolean {
|
|
||||||
return this.hasInitialSyncCompleted;
|
|
||||||
}
|
|
||||||
|
|
||||||
public setHasInitialSyncCompleted(value: boolean): void {
|
|
||||||
this.hasInitialSyncCompleted = value;
|
|
||||||
this.saveInTheBackground();
|
|
||||||
}
|
|
||||||
|
|
||||||
public getLastSeenUpdateId(): VaultUpdateId {
|
|
||||||
return this.lastSeenUpdateIds.min;
|
|
||||||
}
|
|
||||||
|
|
||||||
public addSeenUpdateId(value: number): void {
|
|
||||||
const previousMin = this.lastSeenUpdateIds.min;
|
|
||||||
this.lastSeenUpdateIds.add(value);
|
|
||||||
if (previousMin !== this.lastSeenUpdateIds.min) {
|
|
||||||
this.saveInTheBackground();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
public setLastSeenUpdateId(value: number): void {
|
|
||||||
this.lastSeenUpdateIds.min = value;
|
|
||||||
this.saveInTheBackground();
|
|
||||||
}
|
|
||||||
|
|
||||||
public reset(): void {
|
|
||||||
this.documents = [];
|
|
||||||
this.lastSeenUpdateIds = new CoveredValues(
|
|
||||||
0 // the first updateId will be 1 which is the first integer after -1
|
|
||||||
);
|
|
||||||
this.hasInitialSyncCompleted = false;
|
|
||||||
this.saveInTheBackground();
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public async save(): Promise<void> {
|
public async save(): Promise<void> {
|
||||||
return this.saveData({
|
await this.commit(this.state);
|
||||||
documents: this.resolvedDocuments.map(
|
|
||||||
({ relativePath, documentId, metadata }) => ({
|
|
||||||
documentId,
|
|
||||||
relativePath,
|
|
||||||
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
|
|
||||||
...metadata! // `resolvedDocuments` only returns docs with metadata set
|
|
||||||
})
|
|
||||||
),
|
|
||||||
lastSeenUpdateId: this.lastSeenUpdateIds.min,
|
|
||||||
hasInitialSyncCompleted: this.hasInitialSyncCompleted
|
|
||||||
});
|
|
||||||
}
|
}
|
||||||
|
|
||||||
private ensureConsistency(): void {
|
|
||||||
const idToPath = new Map<string, string[]>();
|
|
||||||
|
|
||||||
this.resolvedDocuments.forEach(({ relativePath, documentId }) => {
|
|
||||||
idToPath.set(documentId, [
|
|
||||||
...(idToPath.get(documentId) ?? []),
|
|
||||||
relativePath
|
|
||||||
]);
|
|
||||||
});
|
|
||||||
|
|
||||||
const duplicates = Array.from(idToPath.entries())
|
|
||||||
.filter(([_, paths]) => paths.length > 1)
|
|
||||||
.map(([id, paths]) => `${id} (${paths.join(", ")})`);
|
|
||||||
|
|
||||||
if (duplicates.length > 0) {
|
|
||||||
throw new Error(
|
|
||||||
"Document IDs are not unique, found duplicates: " +
|
|
||||||
duplicates.join("; ")
|
|
||||||
);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private saveInTheBackground(): void {
|
|
||||||
this.ensureConsistency();
|
|
||||||
void this.save().catch((error: unknown) => {
|
|
||||||
this.logger.error(`Error saving data: ${error}`);
|
|
||||||
});
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue