Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions bin/cli/commands/doctor/stateCacheCapability.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ import type WarpStateCachePort from '../../../../src/ports/WarpStateCachePort.ts
import type WarpStateCacheRetentionPort from '../../../../src/ports/WarpStateCacheRetentionPort.ts';
import defaultCodec from '../../../../src/infrastructure/codecs/CborCodec.ts';
import { DEFAULT_COMMIT_MESSAGE_CODEC } from '../../../../src/infrastructure/adapters/TrailerCommitMessageCodecAdapter.ts';
import defaultCrypto from '../../../../src/infrastructure/adapters/NodeCryptoSingleton.ts';
import type RuntimeStorageProviderPort from '../../../../src/ports/RuntimeStorageProviderPort.ts';
import type { DoctorFinding } from './types.ts';
import {
Expand All @@ -18,6 +19,7 @@ export async function resolveStateCache(
const services = await runtimeStorage.createRuntimeStorageServices({
timelineName: graphName,
codec: defaultCodec,
crypto: defaultCrypto,
commitMessageCodec: DEFAULT_COMMIT_MESSAGE_CODEC,
});
return services.stateSnapshots ?? null;
Expand Down
4 changes: 3 additions & 1 deletion bin/cli/commands/trust.ts
Original file line number Diff line number Diff line change
Expand Up @@ -55,12 +55,14 @@ export default async function handleTrust({ options, args }: { options: CliOptio
const { mode, trustPin } = parseTrustArgs(args);
const { persistence, runtimeStorage, createTrustChain } = await createPersistence(options.repo);
const graphName = await resolveGraphName(persistence, options.graph);
const crypto = new WebCryptoAdapter();
const storage = await runtimeStorage.createRuntimeStorageServices({
timelineName: graphName,
codec: defaultCodec,
crypto,
commitMessageCodec: DEFAULT_COMMIT_MESSAGE_CODEC,
});
const trustChain = createTrustChain(new WebCryptoAdapter());
const trustChain = createTrustChain(crypto);
const verifier = new AuditVerifierService({
auditLog: storage.auditLog,
codec: defaultCodec,
Expand Down
2 changes: 2 additions & 0 deletions bin/cli/commands/verify-audit.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import AuditVerifierService from '../../../src/domain/services/audit/AuditVerifierService.ts';
import defaultCodec from '../../../src/infrastructure/codecs/CborCodec.ts';
import { DEFAULT_COMMIT_MESSAGE_CODEC } from '../../../src/infrastructure/adapters/TrailerCommitMessageCodecAdapter.ts';
import defaultCrypto from '../../../src/infrastructure/adapters/NodeCryptoSingleton.ts';
import { EXIT_CODES, parseCommandArgs, getEnvVar } from '../infrastructure.ts';
import { verifyAuditSchema } from '../schemas.ts';
import { createPersistence, resolveGraphName } from '../shared.ts';
Expand Down Expand Up @@ -52,6 +53,7 @@ export default async function handleVerifyAudit({ options, args }: { options: Cl
const storage = await runtimeStorage.createRuntimeStorageServices({
timelineName: graphName,
codec: defaultCodec,
crypto: defaultCrypto,
commitMessageCodec: DEFAULT_COMMIT_MESSAGE_CODEC,
});
const verifier = new AuditVerifierService({
Expand Down
2 changes: 2 additions & 0 deletions scripts/migrations/v17.0.0/openCheckpointMigrationStore.ts
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import defaultCodec from '../../../src/infrastructure/codecs/CborCodec.ts';
import { DEFAULT_COMMIT_MESSAGE_CODEC } from '../../../src/infrastructure/adapters/TrailerCommitMessageCodecAdapter.ts';
import defaultCrypto from '../../../src/infrastructure/adapters/NodeCryptoSingleton.ts';
import type AssetStoragePort from '../../../src/ports/AssetStoragePort.ts';
import type CheckpointStorePort from '../../../src/ports/CheckpointStorePort.ts';
import type RuntimeStorageProviderPort from '../../../src/ports/RuntimeStorageProviderPort.ts';
Expand All @@ -17,6 +18,7 @@ export async function openCheckpointMigrationStore(
const services = await runtimeStorage.createRuntimeStorageServices({
timelineName: graphName,
codec: defaultCodec,
crypto: defaultCrypto,
commitMessageCodec: DEFAULT_COMMIT_MESSAGE_CODEC,
});
return Object.freeze({
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ import { CONTENT_PROPERTY_KEY }
import GitTimelineHistoryAdapter from '../../../../src/infrastructure/adapters/GitTimelineHistoryAdapter.ts';
import GitCasRepositoryAdapter from '../../../../src/infrastructure/adapters/GitCasRepositoryAdapter.ts';
import { DEFAULT_COMMIT_MESSAGE_CODEC } from '../../../../src/infrastructure/adapters/TrailerCommitMessageCodecAdapter.ts';
import defaultCrypto from '../../../../src/infrastructure/adapters/NodeCryptoSingleton.ts';
import defaultCodec from '../../../../src/infrastructure/codecs/CborCodec.ts';
import type AssetStoragePort from '../../../../src/ports/AssetStoragePort.ts';
import { runMigrationGit } from './GitMigrationCommandRunner.ts';
Expand Down Expand Up @@ -147,6 +148,7 @@ class RuntimeContentOidResolver {
const services = await runtimeStorage.createRuntimeStorageServices({
timelineName: 'migration-content',
codec: defaultCodec,
crypto: defaultCrypto,
commitMessageCodec: DEFAULT_COMMIT_MESSAGE_CODEC,
});
return new RuntimeContentOidResolver(
Expand Down
83 changes: 83 additions & 0 deletions src/domain/materialization/MaterializationCoordinate.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,83 @@
import WarpError from '../errors/WarpError.ts';
import { compareStrings } from '../utils/StringComparison.ts';

export type MaterializationFrontierEntry = Readonly<{
writerId: string;
patchSha: string;
}>;

/** Immutable causal coordinate identifying one exact materialization. */
export default class MaterializationCoordinate {
readonly frontierEntries: readonly MaterializationFrontierEntry[];
readonly ceiling: number | null;

constructor(options: {
readonly frontier: Map<string, string>;
readonly ceiling: number | null;
}) {
requireOptions(options);
this.frontierEntries = freezeFrontier(options.frontier);
this.ceiling = requireCeiling(options.ceiling);
Object.freeze(this);
}

frontier(): Map<string, string> {
return new Map(
this.frontierEntries.map((entry) => [entry.writerId, entry.patchSha]),
);
}

equals(other: MaterializationCoordinate | null | undefined): boolean {
if (!(other instanceof MaterializationCoordinate) || this.ceiling !== other.ceiling) {
return false;
}
if (this.frontierEntries.length !== other.frontierEntries.length) {
return false;
}
return this.frontierEntries.every((entry, index) => {
const candidate = other.frontierEntries[index];
return candidate?.writerId === entry.writerId && candidate.patchSha === entry.patchSha;
});
}
}

function freezeFrontier(frontier: Map<string, string>): readonly MaterializationFrontierEntry[] {
if (!(frontier instanceof Map)) {
throw coordinateError('frontier must be a Map');
}
return Object.freeze(
[...frontier.entries()]
.sort(([left], [right]) => compareStrings(left, right))
.map(([writerId, patchSha]) => Object.freeze({
writerId: requireNonEmpty(writerId, 'frontier writerId'),
patchSha: requireNonEmpty(patchSha, 'frontier patchSha'),
})),
);
}

function requireCeiling(ceiling: number | null): number | null {
if (ceiling === null) {
return null;
}
if (!Number.isSafeInteger(ceiling) || ceiling < 0) {
throw coordinateError('ceiling must be a non-negative safe integer or null');
}
return ceiling;
}

function requireNonEmpty(value: string, field: string): string {
if (typeof value !== 'string' || value.length === 0) {
throw coordinateError(`${field} must be a non-empty string`);
}
return value;
}

function requireOptions(options: object): void {
if (options === null || typeof options !== 'object' || Array.isArray(options)) {
throw coordinateError('options must be an object');
}
}

function coordinateError(message: string): WarpError {
return new WarpError(`Materialization coordinate ${message}`, 'E_MATERIALIZATION_COORDINATE');
}
70 changes: 70 additions & 0 deletions src/domain/materialization/MaterializationHandle.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
import WarpError from '../errors/WarpError.ts';
import BundleHandle from '../storage/BundleHandle.ts';
import StorageRetentionWitness from '../storage/StorageRetentionWitness.ts';
import MaterializationCoordinate from './MaterializationCoordinate.ts';
import MaterializationRoots from './MaterializationRoots.ts';

/** Retained immutable locator and causal identity for one materialization. */
export default class MaterializationHandle {
readonly laneName: string;
readonly bundle: BundleHandle;
readonly coordinate: MaterializationCoordinate;
readonly roots: MaterializationRoots;
readonly stateHash: string;
readonly retention: StorageRetentionWitness;

constructor(options: {
readonly laneName: string;
readonly bundle: BundleHandle;
readonly coordinate: MaterializationCoordinate;
readonly roots: MaterializationRoots;
readonly stateHash: string;
readonly retention: StorageRetentionWitness;
}) {
requireOptions(options);
this.laneName = requireNonEmpty(options.laneName, 'laneName');
this.bundle = requireInstance(options.bundle, BundleHandle, 'bundle');
this.coordinate = requireInstance(
options.coordinate,
MaterializationCoordinate,
'coordinate',
);
this.roots = requireInstance(options.roots, MaterializationRoots, 'roots');
this.stateHash = requireNonEmpty(options.stateHash, 'stateHash');
this.retention = requireInstance(
options.retention,
StorageRetentionWitness,
'retention',
);
if (!this.retention.handle.equals(this.bundle)) {
throw handleError('retention witness does not retain the materialization bundle');
}
Object.freeze(this);
}
}

type RuntimeClass<T> = abstract new (...args: never[]) => T;

function requireInstance<T>(value: T, runtimeClass: RuntimeClass<T>, field: string): T {
if (!(value instanceof runtimeClass)) {
throw handleError(`${field} has an invalid runtime identity`);
}
return value;
}

function requireNonEmpty(value: string, field: string): string {
if (typeof value !== 'string' || value.length === 0) {
throw handleError(`${field} must be a non-empty string`);
}
return value;
}

function requireOptions(options: object): void {
if (options === null || typeof options !== 'object' || Array.isArray(options)) {
throw handleError('options must be an object');
}
}

function handleError(message: string): WarpError {
return new WarpError(`Materialization handle ${message}`, 'E_MATERIALIZATION_HANDLE');
}
97 changes: 97 additions & 0 deletions src/domain/materialization/MaterializationRoots.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,97 @@
import WarpError from '../errors/WarpError.ts';
import BundleHandle from '../storage/BundleHandle.ts';

export const MATERIALIZATION_ROOT_NAMES = defineRootNames(
'adjacency',
'edge-alive',
'edge-births',
'frontier',
'node-alive',
'properties',
'provenance-support',
'roaring-indexes',
);

export type MaterializationRootName = (typeof MATERIALIZATION_ROOT_NAMES)[number];
Comment thread
coderabbitai[bot] marked this conversation as resolved.

export type MaterializationRootsOptions = Readonly<{
adjacency: BundleHandle;
edgeAlive: BundleHandle;
edgeBirths: BundleHandle;
frontier: BundleHandle;
nodeAlive: BundleHandle;
properties: BundleHandle;
provenanceSupport: BundleHandle;
roaringIndexes: BundleHandle;
}>;

/** Independently addressable retained roots for one materialized causal chart. */
export default class MaterializationRoots {
private readonly handles: Readonly<Record<MaterializationRootName, BundleHandle>>;
readonly adjacency: BundleHandle;
readonly edgeAlive: BundleHandle;
readonly edgeBirths: BundleHandle;
readonly frontier: BundleHandle;
readonly nodeAlive: BundleHandle;
readonly properties: BundleHandle;
readonly provenanceSupport: BundleHandle;
readonly roaringIndexes: BundleHandle;

constructor(options: MaterializationRootsOptions) {
requireOptions(options);
this.handles = Object.freeze({
adjacency: requireBundle(options.adjacency, 'adjacency'),
'edge-alive': requireBundle(options.edgeAlive, 'edgeAlive'),
'edge-births': requireBundle(options.edgeBirths, 'edgeBirths'),
frontier: requireBundle(options.frontier, 'frontier'),
'node-alive': requireBundle(options.nodeAlive, 'nodeAlive'),
properties: requireBundle(options.properties, 'properties'),
'provenance-support': requireBundle(options.provenanceSupport, 'provenanceSupport'),
'roaring-indexes': requireBundle(options.roaringIndexes, 'roaringIndexes'),
} satisfies Record<MaterializationRootName, BundleHandle>);
this.adjacency = this.handles.adjacency;
this.edgeAlive = this.handles['edge-alive'];
this.edgeBirths = this.handles['edge-births'];
this.frontier = this.handles.frontier;
this.nodeAlive = this.handles['node-alive'];
this.properties = this.handles.properties;
this.provenanceSupport = this.handles['provenance-support'];
this.roaringIndexes = this.handles['roaring-indexes'];
Object.freeze(this);
}

entries(): readonly (readonly [MaterializationRootName, BundleHandle])[] {
return Object.freeze(
MATERIALIZATION_ROOT_NAMES.map((name) => rootEntry(name, this.handles[name])),
);
}
}

function rootEntry(
name: MaterializationRootName,
handle: BundleHandle,
): readonly [MaterializationRootName, BundleHandle] {
return Object.freeze([name, handle]);
}

function defineRootNames<const Names extends readonly string[]>(...names: Names): Names {
Object.freeze(names);
return names;
}

function requireBundle(handle: BundleHandle, field: string): BundleHandle {
if (!(handle instanceof BundleHandle)) {
throw rootsError(`${field} must be a BundleHandle`);
}
return handle;
}

function requireOptions(options: object): void {
if (options === null || typeof options !== 'object' || Array.isArray(options)) {
throw rootsError('options must be an object');
}
}

function rootsError(message: string): WarpError {
return new WarpError(`Materialization roots ${message}`, 'E_MATERIALIZATION_ROOTS');
}
1 change: 1 addition & 0 deletions src/domain/warp/RuntimeHostBoot.ts
Original file line number Diff line number Diff line change
Expand Up @@ -287,6 +287,7 @@ export async function resolveRuntimeHostConstructionOptions(
const storageServices = await resolvedRuntimeStorage.createRuntimeStorageServices({
timelineName: graphName,
codec: resolvedCodec,
crypto: resolvedCrypto,
commitMessageCodec: resolvedCommitMessageCodec,
...(logger === undefined ? {} : { logger }),
});
Expand Down
Loading
Loading