merge: storage binary-blob primitive across all adapters
Adds saveBinaryBlob/loadBinaryBlob/deleteBinaryBlob/getBinaryBlobPath to the StorageAdapter contract and every adapter (FileSystem real path; remote/memory/ browser return null path). Enables zero-copy mmap-able .cidx segments and batch vector I/O. blobPath convention matches @soulcraft/cortex byte-for-byte.
This commit is contained in:
commit
e23361c488
12 changed files with 1552 additions and 0 deletions
|
|
@ -878,6 +878,52 @@ export interface StorageAdapter {
|
|||
details?: Record<string, any>
|
||||
}>
|
||||
|
||||
/**
|
||||
* Persist a raw binary blob under `key`, writing the bytes verbatim with no
|
||||
* JSON envelope and no base64 inflation. Overwrites any existing blob at the
|
||||
* same key. On filesystem-backed adapters the write is atomic (temp + rename).
|
||||
*
|
||||
* This is the zero-copy / mmap-friendly counterpart to the JSON object
|
||||
* primitives — intended for column-store segments and batch vector payloads
|
||||
* where base64-in-JSON would inflate size ~33% and force full materialization.
|
||||
*
|
||||
* @param key - Logical blob key; "/"-separated segments nest under the
|
||||
* adapter's `_blobs/` prefix (e.g. `"graph-lsm/source/sstable-123"`).
|
||||
* @param data - The exact bytes to store.
|
||||
*/
|
||||
saveBinaryBlob(key: string, data: Buffer): Promise<void>
|
||||
|
||||
/**
|
||||
* Load the raw bytes stored under `key`, byte-identical to what was saved, or
|
||||
* `null` if no blob exists at that key.
|
||||
*
|
||||
* @param key - The blob key used when saving.
|
||||
* @returns The blob bytes, or `null` if absent.
|
||||
*/
|
||||
loadBinaryBlob(key: string): Promise<Buffer | null>
|
||||
|
||||
/**
|
||||
* Delete the blob stored under `key`. Missing blobs are ignored, so delete is
|
||||
* idempotent.
|
||||
*
|
||||
* @param key - The blob key to delete.
|
||||
*/
|
||||
deleteBinaryBlob(key: string): Promise<void>
|
||||
|
||||
/**
|
||||
* Resolve `key` to a real local filesystem path that native code can `mmap`
|
||||
* directly, or `null` when this backend has no local file for the blob.
|
||||
*
|
||||
* Filesystem storage returns the on-disk path; remote object stores, in-memory
|
||||
* storage, browser (OPFS) storage, and historical (read-only) storage return
|
||||
* `null`. `null` is correct behavior for those backends, not a fallback —
|
||||
* callers must fall back to {@link loadBinaryBlob} when no path is available.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @returns An absolute local filesystem path, or `null` if none exists.
|
||||
*/
|
||||
getBinaryBlobPath(key: string): string | null
|
||||
|
||||
/**
|
||||
* Save statistics data
|
||||
* @param statistics The statistics data to save
|
||||
|
|
|
|||
|
|
@ -1007,6 +1007,92 @@ export class AzureBlobStorage extends BaseStorage {
|
|||
}
|
||||
}
|
||||
|
||||
// ===========================================================================
|
||||
// Raw binary-blob primitive
|
||||
// ===========================================================================
|
||||
|
||||
/**
|
||||
* Map a blob key to its Azure blob name under the shared `_blobs/` prefix, e.g.
|
||||
* `"graph-lsm/source/sstable-123"` → `"_blobs/graph-lsm/source/sstable-123.bin"`.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @returns The Azure blob name for the blob.
|
||||
* @private
|
||||
*/
|
||||
private blobObjectKey(key: string): string {
|
||||
return `_blobs/${key}.bin`
|
||||
}
|
||||
|
||||
/**
|
||||
* Store a raw binary blob as an Azure block blob, writing the bytes verbatim
|
||||
* with `application/octet-stream` content type. Overwrites any existing blob at
|
||||
* the same key.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @param data - The exact bytes to store.
|
||||
*/
|
||||
public async saveBinaryBlob(key: string, data: Buffer): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
await this.ensureValidatedForWrite()
|
||||
|
||||
const blockBlobClient = this.containerClient!.getBlockBlobClient(this.blobObjectKey(key))
|
||||
await blockBlobClient.upload(data, data.length, {
|
||||
blobHTTPHeaders: { blobContentType: 'application/octet-stream' }
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Load the raw bytes of the Azure blob for `key`, or `null` if it does not
|
||||
* exist.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @returns The blob bytes, or `null` if absent.
|
||||
*/
|
||||
public async loadBinaryBlob(key: string): Promise<Buffer | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
const blockBlobClient = this.containerClient!.getBlockBlobClient(this.blobObjectKey(key))
|
||||
const downloadResponse = await blockBlobClient.download(0)
|
||||
return await this.streamToBuffer(downloadResponse.readableStreamBody!)
|
||||
} catch (error: any) {
|
||||
if (error.statusCode === 404 || error.code === 'BlobNotFound') {
|
||||
return null
|
||||
}
|
||||
throw BrainyError.fromError(error, `loadBinaryBlob(${key})`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete the Azure blob for `key`. Missing blobs are ignored.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
*/
|
||||
public async deleteBinaryBlob(key: string): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
const blockBlobClient = this.containerClient!.getBlockBlobClient(this.blobObjectKey(key))
|
||||
await blockBlobClient.delete()
|
||||
} catch (error: any) {
|
||||
if (error.statusCode === 404 || error.code === 'BlobNotFound') {
|
||||
return
|
||||
}
|
||||
throw new Error(`Failed to delete blob ${key}: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Azure Blob Storage is remote with no local filesystem path to mmap, so this
|
||||
* always returns `null`. Callers must use {@link loadBinaryBlob} instead.
|
||||
*
|
||||
* @param _key - The blob key (unused).
|
||||
* @returns Always `null`.
|
||||
*/
|
||||
public getBinaryBlobPath(_key: string): string | null {
|
||||
return null
|
||||
}
|
||||
|
||||
/**
|
||||
* Batch delete multiple blobs from Azure Blob Storage
|
||||
* Deletes up to 256 blobs per batch (Azure limit)
|
||||
|
|
|
|||
|
|
@ -129,6 +129,84 @@ export abstract class BaseStorageAdapter implements StorageAdapter {
|
|||
details?: Record<string, any>
|
||||
}>
|
||||
|
||||
// ===========================================================================
|
||||
// Raw binary-blob primitive
|
||||
// ===========================================================================
|
||||
//
|
||||
// A first-class storage primitive for opaque byte payloads that must NOT be
|
||||
// wrapped in a JSON envelope. The JSON object path base64-encodes binary data,
|
||||
// inflating it ~33% and forcing full in-memory materialization on every read.
|
||||
// Blobs side-step that: bytes are written and read verbatim.
|
||||
//
|
||||
// This unblocks zero-copy, mmap-able column-store segments and batch vector
|
||||
// I/O at billion scale. On filesystem-backed adapters `getBinaryBlobPath`
|
||||
// returns a real local path so native code (Rust) can mmap the file directly.
|
||||
//
|
||||
// Blob key → location convention (shared by every adapter so cross-language
|
||||
// and cross-adapter consumers agree on exactly where a blob lives):
|
||||
//
|
||||
// key location
|
||||
// "graph-lsm/source/sstable-123" "<root>/_blobs/graph-lsm/source/sstable-123.bin"
|
||||
//
|
||||
// i.e. the key's "/"-separated segments are nested under a `_blobs/` prefix and
|
||||
// suffixed with `.bin`. Blobs are deliberately NOT branch-scoped (COW): they
|
||||
// are immutable, content-addressed segments managed by their producer,
|
||||
// mirroring how COW metadata under `_cow/` is also kept global.
|
||||
|
||||
/**
|
||||
* Persist a raw binary blob under `key`, writing the bytes verbatim (no JSON
|
||||
* envelope, no base64). Overwrites any existing blob at the same key. Where the
|
||||
* backend is a real filesystem the write is atomic (temp file + rename) so a
|
||||
* concurrent reader never observes a torn write.
|
||||
*
|
||||
* @param key - Logical blob key. "/"-separated segments map to nested
|
||||
* directories under the adapter's `_blobs/` prefix (e.g.
|
||||
* `"graph-lsm/source/sstable-123"`).
|
||||
* @param data - The exact bytes to store.
|
||||
* @returns Resolves once the blob is durably written.
|
||||
* @example
|
||||
* await storage.saveBinaryBlob('graph-lsm/source/sstable-7', segmentBytes)
|
||||
*/
|
||||
abstract saveBinaryBlob(key: string, data: Buffer): Promise<void>
|
||||
|
||||
/**
|
||||
* Load the raw bytes previously stored under `key`, or `null` if no blob
|
||||
* exists at that key. The returned buffer is byte-identical to what was passed
|
||||
* to {@link saveBinaryBlob}.
|
||||
*
|
||||
* @param key - The blob key used when saving.
|
||||
* @returns The blob bytes, or `null` if absent.
|
||||
* @example
|
||||
* const bytes = await storage.loadBinaryBlob('graph-lsm/source/sstable-7')
|
||||
* if (bytes) decodeSegment(bytes)
|
||||
*/
|
||||
abstract loadBinaryBlob(key: string): Promise<Buffer | null>
|
||||
|
||||
/**
|
||||
* Delete the blob stored under `key`. Missing blobs are ignored (no error) so
|
||||
* delete is idempotent.
|
||||
*
|
||||
* @param key - The blob key to delete.
|
||||
* @returns Resolves once the blob is gone (or was already absent).
|
||||
*/
|
||||
abstract deleteBinaryBlob(key: string): Promise<void>
|
||||
|
||||
/**
|
||||
* Resolve `key` to a real local filesystem path that native code can `mmap`
|
||||
* directly, or `null` when this backend has no local file for the blob.
|
||||
*
|
||||
* Filesystem-backed adapters return the on-disk path (whether or not the file
|
||||
* currently exists — the caller is expected to write before mapping). Remote
|
||||
* object stores (S3, R2, GCS, Azure), in-memory storage, browser storage
|
||||
* (OPFS), and the read-only historical adapter genuinely have no local path
|
||||
* and therefore return `null`. A `null` here is correct behavior, not a
|
||||
* fallback: callers must use {@link loadBinaryBlob} when no path is available.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @returns An absolute local filesystem path, or `null` if none exists.
|
||||
*/
|
||||
abstract getBinaryBlobPath(key: string): string | null
|
||||
|
||||
/**
|
||||
* Get optimal batch configuration for this storage adapter
|
||||
* Override in subclasses to provide storage-specific optimization
|
||||
|
|
|
|||
|
|
@ -89,6 +89,10 @@ export class FileSystemStorage extends BaseStorage {
|
|||
private indexDir!: string // Legacy - for backward compatibility
|
||||
private systemDir!: string
|
||||
private lockDir!: string
|
||||
// Root for raw binary blobs (`<rootDir>/_blobs`). Blobs are stored verbatim
|
||||
// (no JSON envelope, no compression) so native code can mmap them directly via
|
||||
// getBinaryBlobPath(). Set in init() once the path module is loaded.
|
||||
private blobsDir!: string
|
||||
private activeLocks: Set<string> = new Set()
|
||||
private lockTimers: Map<string, NodeJS.Timeout> = new Map() // Track timers for cleanup
|
||||
private allTimers: Set<NodeJS.Timeout> = new Set() // Track all timers for cleanup
|
||||
|
|
@ -213,6 +217,7 @@ export class FileSystemStorage extends BaseStorage {
|
|||
this.indexDir = path.join(this.rootDir, 'indexes')
|
||||
this.systemDir = path.join(this.rootDir, SYSTEM_DIR)
|
||||
this.lockDir = path.join(this.rootDir, 'locks')
|
||||
this.blobsDir = path.join(this.rootDir, '_blobs')
|
||||
|
||||
// Create the root directory if it doesn't exist
|
||||
await this.ensureDirectoryExists(this.rootDir)
|
||||
|
|
@ -242,6 +247,9 @@ export class FileSystemStorage extends BaseStorage {
|
|||
// Create the locks directory if it doesn't exist
|
||||
await this.ensureDirectoryExists(this.lockDir)
|
||||
|
||||
// Create the binary blobs directory if it doesn't exist
|
||||
await this.ensureDirectoryExists(this.blobsDir)
|
||||
|
||||
// Initialize count management
|
||||
this.countsFilePath = path.join(this.systemDir, 'counts.json')
|
||||
await this.initializeCounts()
|
||||
|
|
@ -946,6 +954,91 @@ export class FileSystemStorage extends BaseStorage {
|
|||
}
|
||||
}
|
||||
|
||||
// ===========================================================================
|
||||
// Raw binary-blob primitive (mmap-friendly)
|
||||
// ===========================================================================
|
||||
|
||||
/**
|
||||
* Resolve a blob key to its on-disk path under `<rootDir>/_blobs`.
|
||||
*
|
||||
* The key's "/"-separated segments become nested directories and the file is
|
||||
* suffixed with `.bin`, e.g. `"graph-lsm/source/sstable-123"` →
|
||||
* `<rootDir>/_blobs/graph-lsm/source/sstable-123.bin`. This convention is
|
||||
* shared with cortex's `MmapFileSystemStorage` so native code and the TS layer
|
||||
* agree on exactly where each blob lives.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @returns The absolute on-disk path for the blob.
|
||||
* @private
|
||||
*/
|
||||
private blobPath(key: string): string {
|
||||
// `path` and `blobsDir` are populated in init(). If a caller resolves a blob
|
||||
// path before init() (e.g. native code probing for a mmap target), fall back
|
||||
// to a POSIX-style join off rootDir so this synchronous method never throws.
|
||||
if (path && this.blobsDir) {
|
||||
return path.join(this.blobsDir, ...key.split('/')) + '.bin'
|
||||
}
|
||||
return `${this.rootDir}/_blobs/${key}.bin`
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist a raw binary blob verbatim under `key`, using an atomic
|
||||
* temp-file + rename so concurrent readers never observe a torn write.
|
||||
* Parent directories are created on demand.
|
||||
*
|
||||
* @param key - The blob key (see {@link getBinaryBlobPath} for the convention).
|
||||
* @param data - The exact bytes to store.
|
||||
*/
|
||||
public async saveBinaryBlob(key: string, data: Buffer): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
const filePath = this.blobPath(key)
|
||||
await this.ensureDirectoryExists(path.dirname(filePath))
|
||||
const tmpPath = filePath + '.tmp'
|
||||
await fs.promises.writeFile(tmpPath, data)
|
||||
await fs.promises.rename(tmpPath, filePath)
|
||||
}
|
||||
|
||||
/**
|
||||
* Load the raw bytes stored under `key`, or `null` if the blob does not exist.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @returns The blob bytes, or `null` if absent.
|
||||
*/
|
||||
public async loadBinaryBlob(key: string): Promise<Buffer | null> {
|
||||
await this.ensureInitialized()
|
||||
try {
|
||||
return await fs.promises.readFile(this.blobPath(key))
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete the blob stored under `key`. Missing blobs are ignored.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
*/
|
||||
public async deleteBinaryBlob(key: string): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
try {
|
||||
await fs.promises.unlink(this.blobPath(key))
|
||||
} catch {
|
||||
/* ignore missing files */
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the real on-disk path for `key` so native code can mmap the file
|
||||
* directly. The path is returned whether or not the file currently exists —
|
||||
* callers are expected to write before mapping.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @returns The absolute on-disk path (never `null` for filesystem storage).
|
||||
*/
|
||||
public getBinaryBlobPath(key: string): string | null {
|
||||
return this.blobPath(key)
|
||||
}
|
||||
|
||||
/**
|
||||
* Get multiple metadata objects in batches (CRITICAL: Prevents socket exhaustion)
|
||||
* FileSystem implementation uses controlled concurrency to prevent too many file reads
|
||||
|
|
|
|||
|
|
@ -1036,6 +1036,93 @@ export class GcsStorage extends BaseStorage {
|
|||
}
|
||||
}
|
||||
|
||||
// ===========================================================================
|
||||
// Raw binary-blob primitive
|
||||
// ===========================================================================
|
||||
|
||||
/**
|
||||
* Map a blob key to its GCS object name under the shared `_blobs/` prefix, e.g.
|
||||
* `"graph-lsm/source/sstable-123"` → `"_blobs/graph-lsm/source/sstable-123.bin"`.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @returns The GCS object name for the blob.
|
||||
* @private
|
||||
*/
|
||||
private blobObjectKey(key: string): string {
|
||||
return `_blobs/${key}.bin`
|
||||
}
|
||||
|
||||
/**
|
||||
* Store a raw binary blob as a GCS object, writing the bytes verbatim with
|
||||
* `application/octet-stream` content type. Overwrites any existing blob at the
|
||||
* same key.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @param data - The exact bytes to store.
|
||||
*/
|
||||
public async saveBinaryBlob(key: string, data: Buffer): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
await this.ensureValidatedForWrite()
|
||||
|
||||
const file = this.bucket!.file(this.blobObjectKey(key))
|
||||
await file.save(data, {
|
||||
contentType: 'application/octet-stream',
|
||||
resumable: false
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Load the raw bytes of the GCS blob object for `key`, or `null` if it does not
|
||||
* exist.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @returns The blob bytes, or `null` if absent.
|
||||
*/
|
||||
public async loadBinaryBlob(key: string): Promise<Buffer | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
const file = this.bucket!.file(this.blobObjectKey(key))
|
||||
const [contents] = await file.download()
|
||||
return Buffer.from(contents)
|
||||
} catch (error: any) {
|
||||
if (error.code === 404) {
|
||||
return null
|
||||
}
|
||||
throw BrainyError.fromError(error, `loadBinaryBlob(${key})`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete the GCS blob object for `key`. Missing objects are ignored.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
*/
|
||||
public async deleteBinaryBlob(key: string): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
const file = this.bucket!.file(this.blobObjectKey(key))
|
||||
await file.delete()
|
||||
} catch (error: any) {
|
||||
if (error.code === 404) {
|
||||
return
|
||||
}
|
||||
throw new Error(`Failed to delete blob ${key}: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* GCS is a remote object store with no local filesystem path to mmap, so this
|
||||
* always returns `null`. Callers must use {@link loadBinaryBlob} instead.
|
||||
*
|
||||
* @param _key - The blob key (unused).
|
||||
* @returns Always `null`.
|
||||
*/
|
||||
public getBinaryBlobPath(_key: string): string | null {
|
||||
return null
|
||||
}
|
||||
|
||||
/**
|
||||
* List all objects under a specific prefix in GCS
|
||||
* Primitive operation required by base class
|
||||
|
|
|
|||
|
|
@ -264,6 +264,81 @@ export class HistoricalStorageAdapter extends BaseStorage {
|
|||
)
|
||||
}
|
||||
|
||||
// ===========================================================================
|
||||
// Raw binary-blob primitive (read-only)
|
||||
// ===========================================================================
|
||||
|
||||
/**
|
||||
* WRITE BLOCKED: Historical storage is read-only.
|
||||
*
|
||||
* @param key - The blob key (unused; included for the error message).
|
||||
* @throws Always — historical state is immutable.
|
||||
*/
|
||||
public async saveBinaryBlob(key: string, _data: Buffer): Promise<void> {
|
||||
throw new Error(
|
||||
`Historical storage is read-only. Cannot save binary blob: ${key}`
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Load the raw bytes of a binary blob from the historical commit state by
|
||||
* walking the commit tree for the blob entry at `_blobs/<key>.bin` and reading
|
||||
* its content-addressed bytes verbatim (no JSON decode). Returns `null` if the
|
||||
* blob was not present in this commit.
|
||||
*
|
||||
* @param key - The blob key (same convention as the live adapters).
|
||||
* @returns The blob bytes as stored at this commit, or `null` if absent.
|
||||
*/
|
||||
public async loadBinaryBlob(key: string): Promise<Buffer | null> {
|
||||
try {
|
||||
const { CommitObject } = await import('../cow/CommitObject.js')
|
||||
const { TreeObject } = await import('../cow/TreeObject.js')
|
||||
const { isNullHash } = await import('../cow/constants.js')
|
||||
|
||||
const commit = await CommitObject.read(this.blobStorage!, this.commitId)
|
||||
if (isNullHash(commit.tree)) {
|
||||
return null
|
||||
}
|
||||
|
||||
const tree = await TreeObject.read(this.blobStorage!, commit.tree)
|
||||
const blobName = `_blobs/${key}.bin`
|
||||
|
||||
for await (const entry of TreeObject.walk(this.blobStorage!, tree)) {
|
||||
if (entry.type === 'blob' && entry.name === blobName) {
|
||||
return await this.blobStorage!.read(entry.hash)
|
||||
}
|
||||
}
|
||||
|
||||
return null
|
||||
} catch (error) {
|
||||
// Blob not present in historical state
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* DELETE BLOCKED: Historical storage is read-only.
|
||||
*
|
||||
* @param key - The blob key (unused; included for the error message).
|
||||
* @throws Always — historical state is immutable.
|
||||
*/
|
||||
public async deleteBinaryBlob(key: string): Promise<void> {
|
||||
throw new Error(
|
||||
`Historical storage is read-only. Cannot delete binary blob: ${key}`
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Historical state lives in the content-addressed commit store, not on the
|
||||
* local filesystem, so there is no mmap-able path. Always returns `null`.
|
||||
*
|
||||
* @param _key - The blob key (unused).
|
||||
* @returns Always `null`.
|
||||
*/
|
||||
public getBinaryBlobPath(_key: string): string | null {
|
||||
return null
|
||||
}
|
||||
|
||||
/**
|
||||
* Get storage statistics from historical commit
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -30,6 +30,11 @@ export class MemoryStorage extends BaseStorage {
|
|||
// Unified object store for primitive operations (replaces metadata, nounMetadata, verbMetadata)
|
||||
private objectStore: Map<string, any> = new Map()
|
||||
|
||||
// Raw binary-blob store, keyed by blob key. Holds opaque byte payloads
|
||||
// (column-store segments, batch vectors) verbatim — no JSON envelope. In-memory
|
||||
// storage has no local file, so getBinaryBlobPath() returns null.
|
||||
private blobStore: Map<string, Buffer> = new Map()
|
||||
|
||||
// Backward compatibility aliases
|
||||
private get metadata(): Map<string, any> {
|
||||
return this.objectStore
|
||||
|
|
@ -136,6 +141,54 @@ export class MemoryStorage extends BaseStorage {
|
|||
return paths.sort()
|
||||
}
|
||||
|
||||
// ===========================================================================
|
||||
// Raw binary-blob primitive
|
||||
// ===========================================================================
|
||||
|
||||
/**
|
||||
* Persist a raw binary blob in memory under `key`. A defensive copy of the
|
||||
* bytes is stored so later mutations to the caller's buffer don't corrupt the
|
||||
* stored blob. Overwrites any existing blob at the same key.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @param data - The exact bytes to store.
|
||||
*/
|
||||
public async saveBinaryBlob(key: string, data: Buffer): Promise<void> {
|
||||
this.blobStore.set(key, Buffer.from(data))
|
||||
}
|
||||
|
||||
/**
|
||||
* Load a copy of the bytes stored under `key`, or `null` if absent. A copy is
|
||||
* returned so callers cannot mutate the stored blob in place.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @returns The blob bytes, or `null` if absent.
|
||||
*/
|
||||
public async loadBinaryBlob(key: string): Promise<Buffer | null> {
|
||||
const data = this.blobStore.get(key)
|
||||
return data ? Buffer.from(data) : null
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete the blob stored under `key`. Missing blobs are ignored.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
*/
|
||||
public async deleteBinaryBlob(key: string): Promise<void> {
|
||||
this.blobStore.delete(key)
|
||||
}
|
||||
|
||||
/**
|
||||
* In-memory storage has no local filesystem path to mmap, so this always
|
||||
* returns `null`. Callers must use {@link loadBinaryBlob} instead.
|
||||
*
|
||||
* @param _key - The blob key (unused).
|
||||
* @returns Always `null`.
|
||||
*/
|
||||
public getBinaryBlobPath(_key: string): string | null {
|
||||
return null
|
||||
}
|
||||
|
||||
/**
|
||||
* Get multiple metadata objects in batches (CRITICAL: Prevents socket exhaustion)
|
||||
* Memory storage implementation is simple since all data is already in memory
|
||||
|
|
@ -163,6 +216,7 @@ export class MemoryStorage extends BaseStorage {
|
|||
*/
|
||||
public async clear(): Promise<void> {
|
||||
this.objectStore.clear()
|
||||
this.blobStore.clear()
|
||||
this.statistics = null
|
||||
this.totalNounCount = 0
|
||||
this.totalVerbCount = 0
|
||||
|
|
|
|||
|
|
@ -402,6 +402,122 @@ export class OPFSStorage extends BaseStorage {
|
|||
}
|
||||
}
|
||||
|
||||
// ===========================================================================
|
||||
// Raw binary-blob primitive
|
||||
// ===========================================================================
|
||||
|
||||
/**
|
||||
* Resolve a blob key to its OPFS path under the shared `_blobs/` prefix, e.g.
|
||||
* `"graph-lsm/source/sstable-123"` →
|
||||
* `["_blobs", "graph-lsm", "source", "sstable-123.bin"]` (returned as segments
|
||||
* to navigate the OPFS directory tree). Mirrors the on-disk convention used by
|
||||
* filesystem storage.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @returns Path segments for the blob, including the trailing `.bin` filename.
|
||||
* @private
|
||||
*/
|
||||
private blobPathSegments(key: string): string[] {
|
||||
const parts = key.split('/')
|
||||
const filename = parts.pop()! + '.bin'
|
||||
return ['_blobs', ...parts, filename]
|
||||
}
|
||||
|
||||
/**
|
||||
* Store a raw binary blob in OPFS under `key`, writing the bytes verbatim (no
|
||||
* JSON envelope). Parent directories are created on demand. Overwrites any
|
||||
* existing blob at the same key.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @param data - The exact bytes to store.
|
||||
*/
|
||||
public async saveBinaryBlob(key: string, data: Buffer): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
const segments = this.blobPathSegments(key)
|
||||
const filename = segments.pop()!
|
||||
|
||||
let currentDir = this.rootDir!
|
||||
for (const dirName of segments) {
|
||||
currentDir = await currentDir.getDirectoryHandle(dirName, { create: true })
|
||||
}
|
||||
|
||||
const fileHandle = await currentDir.getFileHandle(filename, { create: true })
|
||||
const writable = await fileHandle.createWritable()
|
||||
// Write a copy of the exact bytes; Uint8Array view keeps it binary-clean.
|
||||
await writable.write(new Uint8Array(data))
|
||||
await writable.close()
|
||||
}
|
||||
|
||||
/**
|
||||
* Load the raw bytes of the OPFS blob for `key`, or `null` if it does not
|
||||
* exist.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @returns The blob bytes, or `null` if absent.
|
||||
*/
|
||||
public async loadBinaryBlob(key: string): Promise<Buffer | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
const segments = this.blobPathSegments(key)
|
||||
const filename = segments.pop()!
|
||||
|
||||
let currentDir = this.rootDir!
|
||||
for (const dirName of segments) {
|
||||
currentDir = await currentDir.getDirectoryHandle(dirName)
|
||||
}
|
||||
|
||||
const fileHandle = await currentDir.getFileHandle(filename)
|
||||
const file = await fileHandle.getFile()
|
||||
const arrayBuffer = await file.arrayBuffer()
|
||||
return Buffer.from(arrayBuffer)
|
||||
} catch (error: any) {
|
||||
if (error.name === 'NotFoundError') {
|
||||
return null
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete the OPFS blob for `key`. Missing blobs are ignored.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
*/
|
||||
public async deleteBinaryBlob(key: string): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
const segments = this.blobPathSegments(key)
|
||||
const filename = segments.pop()!
|
||||
|
||||
let currentDir = this.rootDir!
|
||||
for (const dirName of segments) {
|
||||
currentDir = await currentDir.getDirectoryHandle(dirName)
|
||||
}
|
||||
|
||||
await currentDir.removeEntry(filename)
|
||||
} catch (error: any) {
|
||||
if (error.name === 'NotFoundError') {
|
||||
return
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* OPFS is browser-sandboxed storage with no real local filesystem path to
|
||||
* mmap, so this always returns `null`. Callers must use {@link loadBinaryBlob}
|
||||
* instead.
|
||||
*
|
||||
* @param _key - The blob key (unused).
|
||||
* @returns Always `null`.
|
||||
*/
|
||||
public getBinaryBlobPath(_key: string): string | null {
|
||||
return null
|
||||
}
|
||||
|
||||
/**
|
||||
* Get multiple metadata objects in batches (CRITICAL: Prevents socket exhaustion)
|
||||
* OPFS implementation uses controlled concurrency for file operations
|
||||
|
|
|
|||
|
|
@ -725,6 +725,110 @@ export class R2Storage extends BaseStorage {
|
|||
}
|
||||
}
|
||||
|
||||
// ===========================================================================
|
||||
// Raw binary-blob primitive
|
||||
// ===========================================================================
|
||||
|
||||
/**
|
||||
* Map a blob key to its R2 object key under the shared `_blobs/` prefix, e.g.
|
||||
* `"graph-lsm/source/sstable-123"` → `"_blobs/graph-lsm/source/sstable-123.bin"`.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @returns The R2 object key for the blob.
|
||||
* @private
|
||||
*/
|
||||
private blobObjectKey(key: string): string {
|
||||
return `_blobs/${key}.bin`
|
||||
}
|
||||
|
||||
/**
|
||||
* Store a raw binary blob as an R2 object, writing the bytes verbatim with
|
||||
* `application/octet-stream` content type. Overwrites any existing blob at the
|
||||
* same key.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @param data - The exact bytes to store.
|
||||
*/
|
||||
public async saveBinaryBlob(key: string, data: Buffer): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
const { PutObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
await this.s3Client!.send(
|
||||
new PutObjectCommand({
|
||||
Bucket: this.bucketName,
|
||||
Key: this.blobObjectKey(key),
|
||||
Body: data,
|
||||
ContentType: 'application/octet-stream'
|
||||
})
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Load the raw bytes of the R2 blob object for `key`, or `null` if it does not
|
||||
* exist.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @returns The blob bytes, or `null` if absent.
|
||||
*/
|
||||
public async loadBinaryBlob(key: string): Promise<Buffer | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
const { GetObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
const response = await this.s3Client!.send(
|
||||
new GetObjectCommand({
|
||||
Bucket: this.bucketName,
|
||||
Key: this.blobObjectKey(key)
|
||||
})
|
||||
)
|
||||
if (!response || !response.Body) {
|
||||
return null
|
||||
}
|
||||
const bytes = await response.Body.transformToByteArray()
|
||||
return Buffer.from(bytes)
|
||||
} catch (error: any) {
|
||||
if (error.name === 'NoSuchKey' || error.$metadata?.httpStatusCode === 404) {
|
||||
return null
|
||||
}
|
||||
throw BrainyError.fromError(error, `loadBinaryBlob(${key})`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete the R2 blob object for `key`. Missing objects are ignored.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
*/
|
||||
public async deleteBinaryBlob(key: string): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
const { DeleteObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
await this.s3Client!.send(
|
||||
new DeleteObjectCommand({
|
||||
Bucket: this.bucketName,
|
||||
Key: this.blobObjectKey(key)
|
||||
})
|
||||
)
|
||||
} catch (error: any) {
|
||||
if (error.name === 'NoSuchKey' || error.$metadata?.httpStatusCode === 404) {
|
||||
return
|
||||
}
|
||||
throw new Error(`Failed to delete blob ${key}: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* R2 is a remote object store with no local filesystem path to mmap, so this
|
||||
* always returns `null`. Callers must use {@link loadBinaryBlob} instead.
|
||||
*
|
||||
* @param _key - The blob key (unused).
|
||||
* @returns Always `null`.
|
||||
*/
|
||||
public getBinaryBlobPath(_key: string): string | null {
|
||||
return null
|
||||
}
|
||||
|
||||
/**
|
||||
* List all objects under a specific prefix in R2
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -1961,6 +1961,126 @@ export class S3CompatibleStorage extends BaseStorage {
|
|||
}
|
||||
}
|
||||
|
||||
// ===========================================================================
|
||||
// Raw binary-blob primitive
|
||||
// ===========================================================================
|
||||
|
||||
/**
|
||||
* Map a blob key to its S3 object key under the shared `_blobs/` prefix, e.g.
|
||||
* `"graph-lsm/source/sstable-123"` → `"_blobs/graph-lsm/source/sstable-123.bin"`.
|
||||
* Matches the on-disk convention used by filesystem storage so the same logical
|
||||
* key resolves consistently across backends.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @returns The S3 object key for the blob.
|
||||
* @private
|
||||
*/
|
||||
private blobObjectKey(key: string): string {
|
||||
return `_blobs/${key}.bin`
|
||||
}
|
||||
|
||||
/**
|
||||
* Store a raw binary blob as an S3 object, writing the bytes verbatim with
|
||||
* `application/octet-stream` content type (no JSON envelope, no base64).
|
||||
* Overwrites any existing blob at the same key.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @param data - The exact bytes to store.
|
||||
*/
|
||||
public async saveBinaryBlob(key: string, data: Buffer): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
await this.ensureValidatedForWrite()
|
||||
|
||||
const { PutObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
await this.s3Client!.send(
|
||||
new PutObjectCommand({
|
||||
Bucket: this.bucketName,
|
||||
Key: this.blobObjectKey(key),
|
||||
Body: data,
|
||||
ContentType: 'application/octet-stream'
|
||||
})
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Load the raw bytes of the S3 blob object for `key`, or `null` if it does not
|
||||
* exist.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
* @returns The blob bytes, or `null` if absent.
|
||||
*/
|
||||
public async loadBinaryBlob(key: string): Promise<Buffer | null> {
|
||||
await this.ensureInitialized()
|
||||
|
||||
try {
|
||||
const { GetObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
const response = await this.s3Client!.send(
|
||||
new GetObjectCommand({
|
||||
Bucket: this.bucketName,
|
||||
Key: this.blobObjectKey(key)
|
||||
})
|
||||
)
|
||||
if (!response || !response.Body) {
|
||||
return null
|
||||
}
|
||||
const bytes = await response.Body.transformToByteArray()
|
||||
return Buffer.from(bytes)
|
||||
} catch (error: any) {
|
||||
if (
|
||||
error.name === 'NoSuchKey' ||
|
||||
(error.message &&
|
||||
(error.message.includes('NoSuchKey') ||
|
||||
error.message.includes('not found') ||
|
||||
error.message.includes('does not exist')))
|
||||
) {
|
||||
return null
|
||||
}
|
||||
throw BrainyError.fromError(error, `loadBinaryBlob(${key})`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete the S3 blob object for `key`. Missing objects are ignored.
|
||||
*
|
||||
* @param key - The blob key.
|
||||
*/
|
||||
public async deleteBinaryBlob(key: string): Promise<void> {
|
||||
await this.ensureInitialized()
|
||||
await this.ensureValidatedForWrite()
|
||||
|
||||
try {
|
||||
const { DeleteObjectCommand } = await import('@aws-sdk/client-s3')
|
||||
await this.s3Client!.send(
|
||||
new DeleteObjectCommand({
|
||||
Bucket: this.bucketName,
|
||||
Key: this.blobObjectKey(key)
|
||||
})
|
||||
)
|
||||
} catch (error: any) {
|
||||
if (
|
||||
error.name === 'NoSuchKey' ||
|
||||
(error.message &&
|
||||
(error.message.includes('NoSuchKey') ||
|
||||
error.message.includes('not found') ||
|
||||
error.message.includes('does not exist')))
|
||||
) {
|
||||
return
|
||||
}
|
||||
throw new Error(`Failed to delete blob ${key}: ${error}`)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* S3 is a remote object store with no local filesystem path to mmap, so this
|
||||
* always returns `null`. Callers must use {@link loadBinaryBlob} instead.
|
||||
*
|
||||
* @param _key - The blob key (unused).
|
||||
* @returns Always `null`.
|
||||
*/
|
||||
public getBinaryBlobPath(_key: string): string | null {
|
||||
return null
|
||||
}
|
||||
|
||||
/**
|
||||
* Batch delete multiple objects from S3-compatible storage
|
||||
* Deletes up to 1000 objects per batch (S3 limit)
|
||||
|
|
|
|||
|
|
@ -2123,6 +2123,12 @@ export abstract class BaseStorage extends BaseStorageAdapter {
|
|||
*/
|
||||
protected abstract listObjectsUnderPath(prefix: string): Promise<string[]>
|
||||
|
||||
// Raw binary-blob primitive (saveBinaryBlob/loadBinaryBlob/deleteBinaryBlob/
|
||||
// getBinaryBlobPath) is declared abstract on BaseStorageAdapter — the class
|
||||
// that `implements StorageAdapter` — alongside the other public storage
|
||||
// methods. Each concrete adapter implements it. See BaseStorageAdapter for the
|
||||
// contract and the shared `_blobs/<key>.bin` key→location convention.
|
||||
|
||||
/**
|
||||
* Save metadata to storage (now typed)
|
||||
* Routes to correct location (system or entity) based on key format
|
||||
|
|
|
|||
687
tests/unit/storage/binaryBlob.test.ts
Normal file
687
tests/unit/storage/binaryBlob.test.ts
Normal file
|
|
@ -0,0 +1,687 @@
|
|||
/**
|
||||
* Binary Blob Primitive — Unit Tests
|
||||
*
|
||||
* Verifies the raw binary-blob storage primitive
|
||||
* (saveBinaryBlob/loadBinaryBlob/deleteBinaryBlob/getBinaryBlobPath) across
|
||||
* EVERY storage adapter:
|
||||
* - FileSystemStorage (local, real fs path)
|
||||
* - MemoryStorage (in-memory)
|
||||
* - OPFSStorage (browser, exercised via an in-memory OPFS mock)
|
||||
* - S3CompatibleStorage / R2Storage (AWS SDK, exercised via a fake S3 client)
|
||||
* - GcsStorage (exercised via a fake GCS bucket)
|
||||
* - AzureBlobStorage (exercised via a fake container client)
|
||||
* - HistoricalStorageAdapter (read-only)
|
||||
*
|
||||
* Each adapter is checked for: save→load round-trip (byte identical), overwrite,
|
||||
* delete-then-load→null, load-missing→null, and getBinaryBlobPath behavior (a
|
||||
* usable on-disk path for FileSystem, null everywhere else).
|
||||
*
|
||||
* Cloud adapters are exercised against in-memory fakes that implement only the
|
||||
* narrow client surface the blob methods touch. The fakes drive the REAL adapter
|
||||
* code (key→object-key mapping, command construction, byte handling), so these
|
||||
* are genuine round-trips, not mocked assertions.
|
||||
*/
|
||||
|
||||
import { describe, it, expect, beforeEach, afterEach } from 'vitest'
|
||||
import * as os from 'node:os'
|
||||
import * as fsp from 'node:fs/promises'
|
||||
import * as nodePath from 'node:path'
|
||||
|
||||
import { FileSystemStorage } from '../../../src/storage/adapters/fileSystemStorage.js'
|
||||
import { MemoryStorage } from '../../../src/storage/adapters/memoryStorage.js'
|
||||
import { OPFSStorage } from '../../../src/storage/adapters/opfsStorage.js'
|
||||
import { S3CompatibleStorage } from '../../../src/storage/adapters/s3CompatibleStorage.js'
|
||||
import { R2Storage } from '../../../src/storage/adapters/r2Storage.js'
|
||||
import { GcsStorage } from '../../../src/storage/adapters/gcsStorage.js'
|
||||
import { AzureBlobStorage } from '../../../src/storage/adapters/azureBlobStorage.js'
|
||||
import { HistoricalStorageAdapter } from '../../../src/storage/adapters/historicalStorageAdapter.js'
|
||||
|
||||
// Sample payloads. Include non-UTF8 bytes to prove there is no JSON/text
|
||||
// round-tripping anywhere in the path.
|
||||
const PAYLOAD = Buffer.from([0x00, 0x01, 0xff, 0xfe, 0x42, 0x7a, 0x00, 0x80])
|
||||
const PAYLOAD_2 = Buffer.from([0xde, 0xad, 0xbe, 0xef, 0x00, 0x11])
|
||||
const KEY = 'graph-lsm/source/sstable-123'
|
||||
const KEY_FLAT = 'segment-7'
|
||||
|
||||
/**
|
||||
* Shared behavioral contract every adapter must satisfy. `expectsLocalPath`
|
||||
* distinguishes the filesystem adapter (real path) from everyone else (null).
|
||||
*/
|
||||
function runBlobContract(
|
||||
name: string,
|
||||
makeStorage: () => Promise<any>,
|
||||
expectsLocalPath: boolean
|
||||
) {
|
||||
describe(`${name} — binary blob contract`, () => {
|
||||
let storage: any
|
||||
|
||||
beforeEach(async () => {
|
||||
storage = await makeStorage()
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
try {
|
||||
await storage?.clear?.()
|
||||
} catch {
|
||||
/* best-effort cleanup */
|
||||
}
|
||||
})
|
||||
|
||||
it('round-trips bytes identically (save → load)', async () => {
|
||||
await storage.saveBinaryBlob(KEY, PAYLOAD)
|
||||
const loaded = await storage.loadBinaryBlob(KEY)
|
||||
expect(loaded).not.toBeNull()
|
||||
expect(Buffer.isBuffer(loaded)).toBe(true)
|
||||
expect(loaded!.equals(PAYLOAD)).toBe(true)
|
||||
})
|
||||
|
||||
it('overwrites an existing blob', async () => {
|
||||
await storage.saveBinaryBlob(KEY, PAYLOAD)
|
||||
await storage.saveBinaryBlob(KEY, PAYLOAD_2)
|
||||
const loaded = await storage.loadBinaryBlob(KEY)
|
||||
expect(loaded!.equals(PAYLOAD_2)).toBe(true)
|
||||
expect(loaded!.length).toBe(PAYLOAD_2.length)
|
||||
})
|
||||
|
||||
it('returns null after delete', async () => {
|
||||
await storage.saveBinaryBlob(KEY, PAYLOAD)
|
||||
await storage.deleteBinaryBlob(KEY)
|
||||
const loaded = await storage.loadBinaryBlob(KEY)
|
||||
expect(loaded).toBeNull()
|
||||
})
|
||||
|
||||
it('delete is idempotent for a missing blob', async () => {
|
||||
// Should not throw even though nothing exists at this key.
|
||||
await expect(storage.deleteBinaryBlob('does/not/exist')).resolves.toBeUndefined()
|
||||
})
|
||||
|
||||
it('returns null when loading a missing blob', async () => {
|
||||
const loaded = await storage.loadBinaryBlob('never/written')
|
||||
expect(loaded).toBeNull()
|
||||
})
|
||||
|
||||
it('supports flat (non-nested) keys', async () => {
|
||||
await storage.saveBinaryBlob(KEY_FLAT, PAYLOAD)
|
||||
const loaded = await storage.loadBinaryBlob(KEY_FLAT)
|
||||
expect(loaded!.equals(PAYLOAD)).toBe(true)
|
||||
})
|
||||
|
||||
it('getBinaryBlobPath honors the backend contract', () => {
|
||||
const p = storage.getBinaryBlobPath(KEY)
|
||||
if (expectsLocalPath) {
|
||||
expect(typeof p).toBe('string')
|
||||
expect(p).toContain('_blobs')
|
||||
expect(p!.endsWith('.bin')).toBe(true)
|
||||
} else {
|
||||
expect(p).toBeNull()
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// FileSystemStorage
|
||||
// =============================================================================
|
||||
|
||||
describe('FileSystemStorage', () => {
|
||||
let rootDir: string
|
||||
|
||||
runBlobContract(
|
||||
'FileSystemStorage',
|
||||
async () => {
|
||||
rootDir = await fsp.mkdtemp(nodePath.join(os.tmpdir(), 'brainy-blob-fs-'))
|
||||
const storage = new FileSystemStorage(rootDir)
|
||||
await storage.init()
|
||||
return storage
|
||||
},
|
||||
true
|
||||
)
|
||||
|
||||
it('getBinaryBlobPath returns the real on-disk path and the file lands there', async () => {
|
||||
const dir = await fsp.mkdtemp(nodePath.join(os.tmpdir(), 'brainy-blob-fspath-'))
|
||||
const storage = new FileSystemStorage(dir)
|
||||
await storage.init()
|
||||
|
||||
const key = 'graph-lsm/source/sstable-9'
|
||||
const expectedPath = nodePath.join(dir, '_blobs', 'graph-lsm', 'source', 'sstable-9.bin')
|
||||
|
||||
// Path convention matches cortex byte-for-byte: _blobs/<parts>.bin
|
||||
expect(storage.getBinaryBlobPath(key)).toBe(expectedPath)
|
||||
|
||||
await storage.saveBinaryBlob(key, PAYLOAD)
|
||||
|
||||
// The file must physically exist at exactly the advertised path so native
|
||||
// code can mmap it directly.
|
||||
const onDisk = await fsp.readFile(expectedPath)
|
||||
expect(onDisk.equals(PAYLOAD)).toBe(true)
|
||||
|
||||
await fsp.rm(dir, { recursive: true, force: true })
|
||||
})
|
||||
|
||||
it('writes atomically (no leftover .tmp file)', async () => {
|
||||
const dir = await fsp.mkdtemp(nodePath.join(os.tmpdir(), 'brainy-blob-atomic-'))
|
||||
const storage = new FileSystemStorage(dir)
|
||||
await storage.init()
|
||||
|
||||
const key = 'atomic/seg'
|
||||
await storage.saveBinaryBlob(key, PAYLOAD)
|
||||
|
||||
const blobDir = nodePath.join(dir, '_blobs', 'atomic')
|
||||
const entries = await fsp.readdir(blobDir)
|
||||
expect(entries).toContain('seg.bin')
|
||||
expect(entries.some((e) => e.endsWith('.tmp'))).toBe(false)
|
||||
|
||||
await fsp.rm(dir, { recursive: true, force: true })
|
||||
})
|
||||
})
|
||||
|
||||
// =============================================================================
|
||||
// MemoryStorage
|
||||
// =============================================================================
|
||||
|
||||
describe('MemoryStorage', () => {
|
||||
runBlobContract(
|
||||
'MemoryStorage',
|
||||
async () => {
|
||||
const storage = new MemoryStorage()
|
||||
await storage.init()
|
||||
return storage
|
||||
},
|
||||
false
|
||||
)
|
||||
|
||||
it('stores a defensive copy (mutating the input does not corrupt the blob)', async () => {
|
||||
const storage = new MemoryStorage()
|
||||
await storage.init()
|
||||
|
||||
const input = Buffer.from([1, 2, 3, 4])
|
||||
await storage.saveBinaryBlob('k', input)
|
||||
|
||||
// Mutate the caller's buffer after saving.
|
||||
input[0] = 99
|
||||
|
||||
const loaded = await storage.loadBinaryBlob('k')
|
||||
expect(loaded!.equals(Buffer.from([1, 2, 3, 4]))).toBe(true)
|
||||
|
||||
// And the returned buffer is also a copy — mutating it must not affect storage.
|
||||
loaded![1] = 88
|
||||
const reloaded = await storage.loadBinaryBlob('k')
|
||||
expect(reloaded!.equals(Buffer.from([1, 2, 3, 4]))).toBe(true)
|
||||
})
|
||||
|
||||
it('clear() also drops blobs', async () => {
|
||||
const storage = new MemoryStorage()
|
||||
await storage.init()
|
||||
await storage.saveBinaryBlob('k', PAYLOAD)
|
||||
await storage.clear()
|
||||
expect(await storage.loadBinaryBlob('k')).toBeNull()
|
||||
})
|
||||
})
|
||||
|
||||
// =============================================================================
|
||||
// OPFSStorage (browser) — exercised via an in-memory OPFS mock
|
||||
// =============================================================================
|
||||
|
||||
/**
|
||||
* Minimal in-memory implementation of the FileSystem Access API surface that
|
||||
* OPFSStorage uses (getDirectoryHandle / getFileHandle / createWritable /
|
||||
* getFile / arrayBuffer / text / removeEntry / entries). Sufficient to drive the
|
||||
* adapter's real code paths in Node, where OPFS does not exist.
|
||||
*/
|
||||
function createOPFSMock() {
|
||||
class MockFile {
|
||||
constructor(public bytes: Uint8Array) {}
|
||||
async arrayBuffer(): Promise<ArrayBuffer> {
|
||||
// Return a standalone ArrayBuffer copy.
|
||||
return this.bytes.slice().buffer
|
||||
}
|
||||
async text(): Promise<string> {
|
||||
return Buffer.from(this.bytes).toString('utf-8')
|
||||
}
|
||||
get size(): number {
|
||||
return this.bytes.length
|
||||
}
|
||||
}
|
||||
|
||||
class MockWritable {
|
||||
private chunks: Uint8Array[] = []
|
||||
constructor(private fileHandle: MockFileHandle) {}
|
||||
async write(data: any): Promise<void> {
|
||||
if (typeof data === 'string') {
|
||||
this.chunks.push(new Uint8Array(Buffer.from(data, 'utf-8')))
|
||||
} else if (data instanceof Uint8Array) {
|
||||
this.chunks.push(new Uint8Array(data))
|
||||
} else if (data instanceof ArrayBuffer) {
|
||||
this.chunks.push(new Uint8Array(data))
|
||||
} else if (data && data.buffer) {
|
||||
this.chunks.push(new Uint8Array(data.buffer))
|
||||
} else {
|
||||
this.chunks.push(new Uint8Array(Buffer.from(String(data), 'utf-8')))
|
||||
}
|
||||
}
|
||||
async close(): Promise<void> {
|
||||
const total = this.chunks.reduce((n, c) => n + c.length, 0)
|
||||
const merged = new Uint8Array(total)
|
||||
let off = 0
|
||||
for (const c of this.chunks) {
|
||||
merged.set(c, off)
|
||||
off += c.length
|
||||
}
|
||||
this.fileHandle.bytes = merged
|
||||
}
|
||||
}
|
||||
|
||||
class MockFileHandle {
|
||||
kind = 'file' as const
|
||||
constructor(public name: string, public bytes: Uint8Array = new Uint8Array(0)) {}
|
||||
async createWritable(): Promise<MockWritable> {
|
||||
return new MockWritable(this)
|
||||
}
|
||||
async getFile(): Promise<MockFile> {
|
||||
return new MockFile(this.bytes)
|
||||
}
|
||||
}
|
||||
|
||||
class MockDirHandle {
|
||||
kind = 'directory' as const
|
||||
files = new Map<string, MockFileHandle>()
|
||||
dirs = new Map<string, MockDirHandle>()
|
||||
constructor(public name: string) {}
|
||||
|
||||
async getDirectoryHandle(name: string, opts?: { create?: boolean }): Promise<MockDirHandle> {
|
||||
let d = this.dirs.get(name)
|
||||
if (!d) {
|
||||
if (!opts?.create) {
|
||||
const err: any = new Error(`Directory not found: ${name}`)
|
||||
err.name = 'NotFoundError'
|
||||
throw err
|
||||
}
|
||||
d = new MockDirHandle(name)
|
||||
this.dirs.set(name, d)
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
async getFileHandle(name: string, opts?: { create?: boolean }): Promise<MockFileHandle> {
|
||||
let f = this.files.get(name)
|
||||
if (!f) {
|
||||
if (!opts?.create) {
|
||||
const err: any = new Error(`File not found: ${name}`)
|
||||
err.name = 'NotFoundError'
|
||||
throw err
|
||||
}
|
||||
f = new MockFileHandle(name)
|
||||
this.files.set(name, f)
|
||||
}
|
||||
return f
|
||||
}
|
||||
|
||||
async removeEntry(name: string, opts?: { recursive?: boolean }): Promise<void> {
|
||||
if (this.files.has(name)) {
|
||||
this.files.delete(name)
|
||||
return
|
||||
}
|
||||
if (this.dirs.has(name)) {
|
||||
this.dirs.delete(name)
|
||||
return
|
||||
}
|
||||
const err: any = new Error(`Entry not found: ${name}`)
|
||||
err.name = 'NotFoundError'
|
||||
throw err
|
||||
}
|
||||
|
||||
async *entries(): AsyncIterableIterator<[string, MockFileHandle | MockDirHandle]> {
|
||||
for (const [n, f] of this.files) yield [n, f]
|
||||
for (const [n, d] of this.dirs) yield [n, d]
|
||||
}
|
||||
}
|
||||
|
||||
const root = new MockDirHandle('<root>')
|
||||
const navigatorMock = {
|
||||
storage: {
|
||||
getDirectory: async () => root,
|
||||
persisted: async () => true,
|
||||
persist: async () => true,
|
||||
estimate: async () => ({ usage: 0, quota: 1_000_000 })
|
||||
}
|
||||
}
|
||||
return { navigatorMock }
|
||||
}
|
||||
|
||||
describe('OPFSStorage', () => {
|
||||
let hadNavigator = false
|
||||
let originalNavigatorDescriptor: PropertyDescriptor | undefined
|
||||
|
||||
const installMock = () => {
|
||||
const { navigatorMock } = createOPFSMock()
|
||||
// In Node, globalThis.navigator is a read-only getter, so we must redefine
|
||||
// it rather than assign. Capture the original descriptor to restore later.
|
||||
originalNavigatorDescriptor = Object.getOwnPropertyDescriptor(globalThis, 'navigator')
|
||||
hadNavigator = originalNavigatorDescriptor !== undefined
|
||||
Object.defineProperty(globalThis, 'navigator', {
|
||||
value: navigatorMock,
|
||||
configurable: true,
|
||||
writable: true
|
||||
})
|
||||
}
|
||||
|
||||
const restoreMock = () => {
|
||||
if (hadNavigator && originalNavigatorDescriptor) {
|
||||
Object.defineProperty(globalThis, 'navigator', originalNavigatorDescriptor)
|
||||
} else {
|
||||
delete (globalThis as any).navigator
|
||||
}
|
||||
}
|
||||
|
||||
runBlobContract(
|
||||
'OPFSStorage',
|
||||
async () => {
|
||||
installMock()
|
||||
const storage = new OPFSStorage()
|
||||
await storage.init()
|
||||
return storage
|
||||
},
|
||||
false
|
||||
)
|
||||
|
||||
afterEach(() => {
|
||||
restoreMock()
|
||||
})
|
||||
})
|
||||
|
||||
// =============================================================================
|
||||
// Cloud adapter fakes
|
||||
// =============================================================================
|
||||
|
||||
/**
|
||||
* In-memory fake of the AWS S3 v3 client surface used by the blob methods.
|
||||
* Dispatches on the command class name and reads the command's `.input`.
|
||||
* Used by both S3CompatibleStorage and R2Storage (both speak the AWS SDK).
|
||||
*/
|
||||
function createFakeS3Client() {
|
||||
const store = new Map<string, Buffer>()
|
||||
return {
|
||||
store,
|
||||
async send(command: any): Promise<any> {
|
||||
const type = command?.constructor?.name
|
||||
const input = command?.input ?? {}
|
||||
const key = input.Key as string
|
||||
|
||||
if (type === 'PutObjectCommand') {
|
||||
store.set(key, Buffer.from(input.Body))
|
||||
return {}
|
||||
}
|
||||
if (type === 'GetObjectCommand') {
|
||||
const data = store.get(key)
|
||||
if (!data) {
|
||||
const err: any = new Error('NoSuchKey')
|
||||
err.name = 'NoSuchKey'
|
||||
err.$metadata = { httpStatusCode: 404 }
|
||||
throw err
|
||||
}
|
||||
return {
|
||||
Body: {
|
||||
transformToByteArray: async () => new Uint8Array(data),
|
||||
transformToString: async () => data.toString('utf-8')
|
||||
}
|
||||
}
|
||||
}
|
||||
if (type === 'DeleteObjectCommand') {
|
||||
store.delete(key)
|
||||
return {}
|
||||
}
|
||||
throw new Error(`FakeS3Client: unhandled command ${type}`)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** In-memory fake of the GCS Bucket surface used by the blob methods. */
|
||||
function createFakeGcsBucket() {
|
||||
const store = new Map<string, Buffer>()
|
||||
return {
|
||||
store,
|
||||
file(name: string) {
|
||||
return {
|
||||
async save(data: Buffer): Promise<void> {
|
||||
store.set(name, Buffer.from(data))
|
||||
},
|
||||
async download(): Promise<[Buffer]> {
|
||||
const data = store.get(name)
|
||||
if (!data) {
|
||||
const err: any = new Error('Not Found')
|
||||
err.code = 404
|
||||
throw err
|
||||
}
|
||||
return [Buffer.from(data)]
|
||||
},
|
||||
async delete(): Promise<void> {
|
||||
if (!store.has(name)) {
|
||||
const err: any = new Error('Not Found')
|
||||
err.code = 404
|
||||
throw err
|
||||
}
|
||||
store.delete(name)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** In-memory fake of the Azure ContainerClient surface used by the blob methods. */
|
||||
function createFakeAzureContainerClient() {
|
||||
const store = new Map<string, Buffer>()
|
||||
return {
|
||||
store,
|
||||
getBlockBlobClient(name: string) {
|
||||
return {
|
||||
async upload(data: Buffer, _length: number): Promise<void> {
|
||||
store.set(name, Buffer.from(data))
|
||||
},
|
||||
async download(_offset: number): Promise<any> {
|
||||
const data = store.get(name)
|
||||
if (!data) {
|
||||
const err: any = new Error('BlobNotFound')
|
||||
err.statusCode = 404
|
||||
err.code = 'BlobNotFound'
|
||||
throw err
|
||||
}
|
||||
// Provide a Node Readable stream of the bytes; streamToBuffer consumes it.
|
||||
const { Readable } = require('node:stream')
|
||||
return { readableStreamBody: Readable.from([data]) }
|
||||
},
|
||||
async delete(): Promise<void> {
|
||||
if (!store.has(name)) {
|
||||
const err: any = new Error('BlobNotFound')
|
||||
err.statusCode = 404
|
||||
err.code = 'BlobNotFound'
|
||||
throw err
|
||||
}
|
||||
store.delete(name)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// S3CompatibleStorage (fake client)
|
||||
// =============================================================================
|
||||
|
||||
describe('S3CompatibleStorage', () => {
|
||||
runBlobContract(
|
||||
'S3CompatibleStorage',
|
||||
async () => {
|
||||
const storage = new S3CompatibleStorage({
|
||||
bucketName: 'test-bucket',
|
||||
region: 'us-east-1',
|
||||
accessKeyId: 'test',
|
||||
secretAccessKey: 'test'
|
||||
})
|
||||
// Bypass real network init; inject in-memory fake client.
|
||||
;(storage as any).isInitialized = true
|
||||
;(storage as any).bucketValidated = true
|
||||
;(storage as any).s3Client = createFakeS3Client()
|
||||
return storage
|
||||
},
|
||||
false
|
||||
)
|
||||
|
||||
it('maps the blob key to the _blobs/<key>.bin object key', async () => {
|
||||
const storage = new S3CompatibleStorage({
|
||||
bucketName: 'test-bucket',
|
||||
region: 'us-east-1',
|
||||
accessKeyId: 'test',
|
||||
secretAccessKey: 'test'
|
||||
})
|
||||
const fake = createFakeS3Client()
|
||||
;(storage as any).isInitialized = true
|
||||
;(storage as any).bucketValidated = true
|
||||
;(storage as any).s3Client = fake
|
||||
|
||||
await storage.saveBinaryBlob('graph-lsm/source/sstable-1', PAYLOAD)
|
||||
expect([...fake.store.keys()]).toEqual(['_blobs/graph-lsm/source/sstable-1.bin'])
|
||||
})
|
||||
})
|
||||
|
||||
// =============================================================================
|
||||
// R2Storage (fake client)
|
||||
// =============================================================================
|
||||
|
||||
describe('R2Storage', () => {
|
||||
runBlobContract(
|
||||
'R2Storage',
|
||||
async () => {
|
||||
const storage = new R2Storage({
|
||||
bucketName: 'test-bucket',
|
||||
accountId: 'acct123',
|
||||
accessKeyId: 'test',
|
||||
secretAccessKey: 'test'
|
||||
})
|
||||
;(storage as any).isInitialized = true
|
||||
;(storage as any).bucketValidated = true
|
||||
;(storage as any).s3Client = createFakeS3Client()
|
||||
return storage
|
||||
},
|
||||
false
|
||||
)
|
||||
})
|
||||
|
||||
// =============================================================================
|
||||
// GcsStorage (fake bucket)
|
||||
// =============================================================================
|
||||
|
||||
describe('GcsStorage', () => {
|
||||
runBlobContract(
|
||||
'GcsStorage',
|
||||
async () => {
|
||||
const storage = new GcsStorage({ bucketName: 'test-bucket' })
|
||||
;(storage as any).isInitialized = true
|
||||
;(storage as any).bucketValidated = true
|
||||
;(storage as any).bucket = createFakeGcsBucket()
|
||||
return storage
|
||||
},
|
||||
false
|
||||
)
|
||||
})
|
||||
|
||||
// =============================================================================
|
||||
// AzureBlobStorage (fake container client)
|
||||
// =============================================================================
|
||||
|
||||
describe('AzureBlobStorage', () => {
|
||||
runBlobContract(
|
||||
'AzureBlobStorage',
|
||||
async () => {
|
||||
const storage = new AzureBlobStorage({
|
||||
containerName: 'test-container',
|
||||
connectionString:
|
||||
'DefaultEndpointsProtocol=https;AccountName=test;AccountKey=dGVzdA==;EndpointSuffix=core.windows.net'
|
||||
})
|
||||
;(storage as any).isInitialized = true
|
||||
;(storage as any).bucketValidated = true
|
||||
;(storage as any).containerClient = createFakeAzureContainerClient()
|
||||
return storage
|
||||
},
|
||||
false
|
||||
)
|
||||
})
|
||||
|
||||
// =============================================================================
|
||||
// HistoricalStorageAdapter (read-only)
|
||||
// =============================================================================
|
||||
|
||||
describe('HistoricalStorageAdapter — binary blob (read-only)', () => {
|
||||
/**
|
||||
* Build a HistoricalStorageAdapter over a real COW-enabled MemoryStorage, then
|
||||
* commit a binary blob into the tree so loadBinaryBlob can resolve it from the
|
||||
* historical commit state.
|
||||
*/
|
||||
async function makeHistorical(withBlob: boolean) {
|
||||
const underlying = new MemoryStorage()
|
||||
await underlying.init()
|
||||
await underlying.initializeCOW({ branch: 'main' })
|
||||
|
||||
const blobStorage = underlying.blobStorage!
|
||||
const refManager = underlying.refManager!
|
||||
|
||||
const { TreeBuilder } = await import('../../../src/storage/cow/TreeObject.js')
|
||||
const { CommitBuilder } = await import('../../../src/storage/cow/CommitObject.js')
|
||||
|
||||
// Build a tree; optionally include a _blobs/<key>.bin entry holding raw bytes.
|
||||
const treeBuilder = TreeBuilder.create(blobStorage)
|
||||
if (withBlob) {
|
||||
const blobHash = await blobStorage.write(PAYLOAD)
|
||||
treeBuilder.addBlob('_blobs/graph-lsm/source/sstable-123.bin', blobHash, PAYLOAD.length)
|
||||
}
|
||||
const treeHash = await treeBuilder.build()
|
||||
|
||||
const commitHash = await CommitBuilder.create(blobStorage)
|
||||
.tree(treeHash)
|
||||
.parent(null)
|
||||
.message('blob commit')
|
||||
.author('test')
|
||||
.timestamp(Date.now())
|
||||
.build()
|
||||
|
||||
await refManager.createBranch('snapshot', commitHash, {
|
||||
description: 'snapshot',
|
||||
author: 'test'
|
||||
})
|
||||
|
||||
const historical = new HistoricalStorageAdapter({
|
||||
underlyingStorage: underlying,
|
||||
commitId: commitHash,
|
||||
branch: 'main'
|
||||
})
|
||||
await historical.init()
|
||||
return historical
|
||||
}
|
||||
|
||||
it('loads a blob committed into the historical state (bytes identical)', async () => {
|
||||
const historical = await makeHistorical(true)
|
||||
const loaded = await historical.loadBinaryBlob('graph-lsm/source/sstable-123')
|
||||
expect(loaded).not.toBeNull()
|
||||
expect(loaded!.equals(PAYLOAD)).toBe(true)
|
||||
})
|
||||
|
||||
it('returns null for a blob absent from the historical state', async () => {
|
||||
const historical = await makeHistorical(false)
|
||||
const loaded = await historical.loadBinaryBlob('graph-lsm/source/sstable-123')
|
||||
expect(loaded).toBeNull()
|
||||
})
|
||||
|
||||
it('saveBinaryBlob throws (read-only)', async () => {
|
||||
const historical = await makeHistorical(false)
|
||||
await expect(historical.saveBinaryBlob('k', PAYLOAD)).rejects.toThrow(/read-only/i)
|
||||
})
|
||||
|
||||
it('deleteBinaryBlob throws (read-only)', async () => {
|
||||
const historical = await makeHistorical(false)
|
||||
await expect(historical.deleteBinaryBlob('k')).rejects.toThrow(/read-only/i)
|
||||
})
|
||||
|
||||
it('getBinaryBlobPath returns null', async () => {
|
||||
const historical = await makeHistorical(false)
|
||||
expect(historical.getBinaryBlobPath('k')).toBeNull()
|
||||
})
|
||||
})
|
||||
Loading…
Add table
Add a link
Reference in a new issue